save

Dockerfile

@@ -4,6 +4,8 @@ ENV TZ=Asia/Shanghai
 RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && echo $TZ > /etc/timezone
 RUN mkdir -p $FLINK_HOME/usrlib
 RUN mkdir -p /conf
+RUN mkdir -p /opt/flink/plugins/s3-fs-presto
+RUN cp /opt/flink/opt/flink-s3-fs-presto-1.19.1.jar /opt/flink/plugins/s3-fs-presto/
 
 #ENV HADOOP_CLASSPATH="`hadoop classpath`"
 COPY ./opentelemetry-javaagent.jar /opentelemetry-javaagent.jar

src/main/java/com/cecf/observe/agentstream/Main.java

@@ -59,8 +59,10 @@ public class Main {
         WatermarkStrategy<PojoSpan> watermarkStrategy = WatermarkStrategy
                 .<PojoSpan>forBoundedOutOfOrderness(Duration.ofSeconds(jobConfig.getWatermarkOrderlessSeconds()))
                 .withTimestampAssigner(
-                        (SerializableTimestampAssigner<PojoSpan>) (element, recordTimestamp) -> element.Timestamp.getTime()
-                ).withIdleness(Duration.ofSeconds(jobConfig.getWatermarkIdlenessSeconds()));
+                        (SerializableTimestampAssigner<PojoSpan>) (element, recordTimestamp) -> recordTimestamp
+                ).withIdleness(
+                        Duration.ofSeconds(jobConfig.getWatermarkIdlenessSeconds())
+                );
         if (enablePbSource) {
             OasConfig.KafkaSourceConfig pbKafkaSourceConfig = oasConfig.getPbSource();
             DeserializationSchema<ExportTraceServiceRequest> pbDeserializationSchema = new ExportTraceServiceRequestSchema();

src/main/java/com/cecf/observe/agentstream/OasConfig.java

@@ -13,6 +13,14 @@ import java.util.Map;
 @Setter
 @Getter
 public class OasConfig {
+    @ToString
+    @Getter
+    @Setter
+    public static class SlotShareGroupConfig {
+        private Integer cpu;
+        private Integer memoryMB;
+    }
+
     @ToString
     @Getter
     @Setter

version

@@ -1 +1 @@
-v2.3.2
+v2.3.4