fix: convert AbstractColumnTimestampExtractor to KsqlTimestampExtractor

confluentinc · Feb 25, 2020 · 2eece42 · 2eece42
1 parent dc5c900
commit 2eece42
Show file tree

Hide file tree

Showing 25 changed files with 233 additions and 160 deletions.
diff --git a/...expected_topology/0_6_0-pre/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts b/...expected_topology/0_6_0-pre/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts
@@ -79,11 +79,11 @@ Topologies:
       --> Project
       <-- Join-this-join, Join-other-join
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000012
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Join-merge
-    Processor: KSTREAM-TRANSFORM-0000000012 (stores: [])
-      --> KSTREAM-SINK-0000000013
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000012
       <-- Project
-    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000012
+    Sink: KSTREAM-SINK-0000000012 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/..._0-pre/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts_extractor_both_sides b/..._0-pre/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts_extractor_both_sides
@@ -79,11 +79,11 @@ Topologies:
       --> Project
       <-- Join-this-join, Join-other-join
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000012
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Join-merge
-    Processor: KSTREAM-TRANSFORM-0000000012 (stores: [])
-      --> KSTREAM-SINK-0000000013
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000012
       <-- Project
-    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000012
+    Sink: KSTREAM-SINK-0000000012 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/...ogy/0_6_0-pre/join-with-custom-timestamp_-_stream_table_join_with_ts_extractor_both_sides b/...ogy/0_6_0-pre/join-with-custom-timestamp_-_stream_table_join_with_ts_extractor_both_sides
@@ -63,16 +63,16 @@ Topologies:
       --> KTABLE-TRANSFORMVALUES-0000000002
       <-- KSTREAM-SOURCE-0000000000
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000009
+      --> ApplyTimestampTransform-S1_JOIN_T1
       <-- Join
-    Processor: KSTREAM-TRANSFORM-0000000009 (stores: [])
-      --> KSTREAM-SINK-0000000010
+    Processor: ApplyTimestampTransform-S1_JOIN_T1 (stores: [])
+      --> KSTREAM-SINK-0000000009
       <-- Project
     Processor: KTABLE-TRANSFORMVALUES-0000000002 (stores: [])
       --> PrependAliasRight
       <-- KTABLE-SOURCE-0000000001
-    Sink: KSTREAM-SINK-0000000010 (topic: S1_JOIN_T1)
-      <-- KSTREAM-TRANSFORM-0000000009
+    Sink: KSTREAM-SINK-0000000009 (topic: S1_JOIN_T1)
+      <-- ApplyTimestampTransform-S1_JOIN_T1
     Processor: PrependAliasRight (stores: [])
       --> none
       <-- KTABLE-TRANSFORMVALUES-0000000002

diff --git a/...s/expected_topology/0_6_0-pre/join-with-custom-timestamp_-_table_table_inner_join_with_ts b/...s/expected_topology/0_6_0-pre/join-with-custom-timestamp_-_table_table_inner_join_with_ts
@@ -80,11 +80,11 @@ Topologies:
       --> KTABLE-TOSTREAM-0000000012
       <-- KTABLE-MERGE-0000000008
     Processor: KTABLE-TOSTREAM-0000000012 (stores: [])
-      --> KSTREAM-TRANSFORM-0000000013
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Project
-    Processor: KSTREAM-TRANSFORM-0000000013 (stores: [])
-      --> KSTREAM-SINK-0000000014
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000013
       <-- KTABLE-TOSTREAM-0000000012
-    Sink: KSTREAM-SINK-0000000014 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000013
+    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/..._6_0-pre/join-with-custom-timestamp_-_table_table_inner_join_with_ts_extractor_both_sides b/..._6_0-pre/join-with-custom-timestamp_-_table_table_inner_join_with_ts_extractor_both_sides
@@ -80,11 +80,11 @@ Topologies:
       --> KTABLE-TOSTREAM-0000000012
       <-- KTABLE-MERGE-0000000008
     Processor: KTABLE-TOSTREAM-0000000012 (stores: [])
-      --> KSTREAM-TRANSFORM-0000000013
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Project
-    Processor: KSTREAM-TRANSFORM-0000000013 (stores: [])
-      --> KSTREAM-SINK-0000000014
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000013
       <-- KTABLE-TOSTREAM-0000000012
-    Sink: KSTREAM-SINK-0000000014 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000013
+    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/...ces/expected_topology/0_7_0/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts b/...ces/expected_topology/0_7_0/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts
@@ -83,11 +83,11 @@ Topologies:
       --> Project
       <-- Join-this-join, Join-other-join
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000012
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Join-merge
-    Processor: KSTREAM-TRANSFORM-0000000012 (stores: [])
-      --> KSTREAM-SINK-0000000013
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000012
       <-- Project
-    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000012
+    Sink: KSTREAM-SINK-0000000012 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/.../0_7_0/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts_extractor_both_sides b/.../0_7_0/join-with-custom-timestamp_-_stream_stream_inner_join_with_ts_extractor_both_sides
@@ -83,11 +83,11 @@ Topologies:
       --> Project
       <-- Join-this-join, Join-other-join
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000012
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Join-merge
-    Processor: KSTREAM-TRANSFORM-0000000012 (stores: [])
-      --> KSTREAM-SINK-0000000013
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000012
       <-- Project
-    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000012
+    Sink: KSTREAM-SINK-0000000012 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/...opology/0_7_0/join-with-custom-timestamp_-_stream_table_join_with_ts_extractor_both_sides b/...opology/0_7_0/join-with-custom-timestamp_-_stream_table_join_with_ts_extractor_both_sides
@@ -67,16 +67,16 @@ Topologies:
       --> KTABLE-TRANSFORMVALUES-0000000002
       <-- KSTREAM-SOURCE-0000000000
     Processor: Project (stores: [])
-      --> KSTREAM-TRANSFORM-0000000009
+      --> ApplyTimestampTransform-S1_JOIN_T1
       <-- Join
-    Processor: KSTREAM-TRANSFORM-0000000009 (stores: [])
-      --> KSTREAM-SINK-0000000010
+    Processor: ApplyTimestampTransform-S1_JOIN_T1 (stores: [])
+      --> KSTREAM-SINK-0000000009
       <-- Project
     Processor: KTABLE-TRANSFORMVALUES-0000000002 (stores: [])
       --> PrependAliasRight
       <-- KTABLE-SOURCE-0000000001
-    Sink: KSTREAM-SINK-0000000010 (topic: S1_JOIN_T1)
-      <-- KSTREAM-TRANSFORM-0000000009
+    Sink: KSTREAM-SINK-0000000009 (topic: S1_JOIN_T1)
+      <-- ApplyTimestampTransform-S1_JOIN_T1
     Processor: PrependAliasRight (stores: [])
       --> none
       <-- KTABLE-TRANSFORMVALUES-0000000002

diff --git a/...urces/expected_topology/0_7_0/join-with-custom-timestamp_-_table_table_inner_join_with_ts b/...urces/expected_topology/0_7_0/join-with-custom-timestamp_-_table_table_inner_join_with_ts
@@ -84,11 +84,11 @@ Topologies:
       --> KTABLE-TOSTREAM-0000000012
       <-- KTABLE-MERGE-0000000008
     Processor: KTABLE-TOSTREAM-0000000012 (stores: [])
-      --> KSTREAM-TRANSFORM-0000000013
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Project
-    Processor: KSTREAM-TRANSFORM-0000000013 (stores: [])
-      --> KSTREAM-SINK-0000000014
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000013
       <-- KTABLE-TOSTREAM-0000000012
-    Sink: KSTREAM-SINK-0000000014 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000013
+    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/...gy/0_7_0/join-with-custom-timestamp_-_table_table_inner_join_with_ts_extractor_both_sides b/...gy/0_7_0/join-with-custom-timestamp_-_table_table_inner_join_with_ts_extractor_both_sides
@@ -84,11 +84,11 @@ Topologies:
       --> KTABLE-TOSTREAM-0000000012
       <-- KTABLE-MERGE-0000000008
     Processor: KTABLE-TOSTREAM-0000000012 (stores: [])
-      --> KSTREAM-TRANSFORM-0000000013
+      --> ApplyTimestampTransform-S1_JOIN_S2
       <-- Project
-    Processor: KSTREAM-TRANSFORM-0000000013 (stores: [])
-      --> KSTREAM-SINK-0000000014
+    Processor: ApplyTimestampTransform-S1_JOIN_S2 (stores: [])
+      --> KSTREAM-SINK-0000000013
       <-- KTABLE-TOSTREAM-0000000012
-    Sink: KSTREAM-SINK-0000000014 (topic: S1_JOIN_S2)
-      <-- KSTREAM-TRANSFORM-0000000013
+    Sink: KSTREAM-SINK-0000000013 (topic: S1_JOIN_S2)
+      <-- ApplyTimestampTransform-S1_JOIN_S2
 
diff --git a/ksql-rest-app/src/test/resources/ksql-plan-schema/schema.json b/ksql-rest-app/src/test/resources/ksql-plan-schema/schema.json
@@ -453,7 +453,7 @@
         }
       },
       "title" : "streamSinkV1",
-      "required" : [ "@type", "properties", "source", "formats", "topicName", "timestampColumn" ]
+      "required" : [ "@type", "properties", "source", "formats", "topicName" ]
     },
     "StreamSource" : {
       "type" : "object",
@@ -812,7 +812,7 @@
         }
       },
       "title" : "tableSinkV1",
-      "required" : [ "@type", "properties", "source", "formats", "topicName", "timestampColumn" ]
+      "required" : [ "@type", "properties", "source", "formats", "topicName" ]
     },
     "TableTableJoin" : {
       "type" : "object",

diff --git a/ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java b/ksql-streams/src/main/java/io/confluent/ksql/execution/streams/SinkBuilder.java
@@ -22,7 +22,7 @@
 import io.confluent.ksql.execution.context.QueryLoggerUtil;
 import io.confluent.ksql.execution.plan.Formats;
 import io.confluent.ksql.execution.plan.KeySerdeFactory;
-import io.confluent.ksql.execution.streams.timestamp.AbstractColumnTimestampExtractor;
+import io.confluent.ksql.execution.streams.timestamp.KsqlTimestampExtractor;
 import io.confluent.ksql.execution.streams.timestamp.TimestampExtractionPolicy;
 import io.confluent.ksql.execution.streams.timestamp.TimestampExtractionPolicyFactory;
 import io.confluent.ksql.execution.timestamp.TimestampColumn;
@@ -43,6 +43,8 @@
 import org.apache.kafka.streams.processor.To;
 
 public final class SinkBuilder {
+  private static final String TIMESTAMP_TRANSFORM_NAME = "ApplyTimestampTransform-";
+
   private SinkBuilder() {
   }
 
@@ -78,7 +80,8 @@ public static  <K> void build(
     );
 
     final KStream<K, GenericRow> transformed = tsTransformer
-        .map(t -> stream.transform(t, Named.as(StreamsUtil.buildOpName(queryContext))))
+        .map(t -> stream.transform(t, Named.as(TIMESTAMP_TRANSFORM_NAME
+            + StreamsUtil.buildOpName(queryContext))))
         .orElse(stream);
 
     transformed.to(topicName, Produced.with(keySerde, valueSerde));
@@ -105,9 +108,8 @@ private static  <K> Optional<TransformTimestamp<K>> timestampTransformer(
         .map(c -> sourceSchema.findValueColumn(c).orElseThrow(IllegalStateException::new))
         .map(Column::index)
         .map(timestampPolicy::create)
-        .filter(te -> te instanceof AbstractColumnTimestampExtractor)
         .map(te -> new TransformTimestamp<>(
-            (AbstractColumnTimestampExtractor)te,
+            te,
             queryBuilder
                 .getProcessingLogContext()
                 .getLoggerFactory()
@@ -123,11 +125,11 @@ private static  <K> Optional<TransformTimestamp<K>> timestampTransformer(
 
   static class TransformTimestamp<K>
       implements TransformerSupplier<K, GenericRow, KeyValue<K, GenericRow>> {
-    private final AbstractColumnTimestampExtractor timestampExtractor;
+    private final KsqlTimestampExtractor timestampExtractor;
     private final ProcessingLogger processingLogger;
 
     TransformTimestamp(
-        final AbstractColumnTimestampExtractor timestampExtractor,
+        final KsqlTimestampExtractor timestampExtractor,
         final ProcessingLogger processingLogger
     ) {
       this.timestampExtractor = requireNonNull(timestampExtractor, "timestampExtractor");

diff --git a/.../java/io/confluent/ksql/execution/streams/timestamp/AbstractColumnTimestampExtractor.java b/.../java/io/confluent/ksql/execution/streams/timestamp/AbstractColumnTimestampExtractor.java
diff --git a/...s/src/main/java/io/confluent/ksql/execution/streams/timestamp/KsqlTimestampExtractor.java b/...s/src/main/java/io/confluent/ksql/execution/streams/timestamp/KsqlTimestampExtractor.java
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2020 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.ksql.execution.streams.timestamp;
+
+import io.confluent.ksql.GenericRow;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.streams.processor.TimestampExtractor;
+
+public interface KsqlTimestampExtractor extends TimestampExtractor {
+  default long extract(final ConsumerRecord<Object, Object> record, final long previousTimestamp) {
+    return extract((GenericRow) record.value());
+  }
+
+  long extract(GenericRow row);
+}
diff --git a/...va/io/confluent/ksql/execution/streams/timestamp/LongColumnTimestampExtractionPolicy.java b/...va/io/confluent/ksql/execution/streams/timestamp/LongColumnTimestampExtractionPolicy.java
@@ -18,7 +18,6 @@
 import com.google.errorprone.annotations.Immutable;
 import io.confluent.ksql.name.ColumnName;
 import java.util.Objects;
-import org.apache.kafka.streams.processor.TimestampExtractor;
 
 @Immutable
 public class LongColumnTimestampExtractionPolicy implements TimestampExtractionPolicy {
@@ -31,7 +30,7 @@ public LongColumnTimestampExtractionPolicy(final ColumnName timestampField) {
   }
 
   @Override
-  public TimestampExtractor create(final int columnIndex) {
+  public KsqlTimestampExtractor create(final int columnIndex) {
     return new LongTimestampExtractor(columnIndex);
   }
 

diff --git a/...s/src/main/java/io/confluent/ksql/execution/streams/timestamp/LongTimestampExtractor.java b/...s/src/main/java/io/confluent/ksql/execution/streams/timestamp/LongTimestampExtractor.java
@@ -17,13 +17,15 @@
 
 import io.confluent.ksql.GenericRow;
 
-public class LongTimestampExtractor extends AbstractColumnTimestampExtractor {
+public class LongTimestampExtractor implements KsqlTimestampExtractor {
+  private final int timestampColumnindex;
+
   LongTimestampExtractor(final int timestampColumnindex) {
-    super(timestampColumnindex);
+    this.timestampColumnindex = timestampColumnindex;
   }
 
   @Override
   public long extract(final GenericRow row) {
-    return (long)row.get(timetampColumnIndex);
+    return (long)row.get(timestampColumnindex);
   }
 }
diff --git a/...java/io/confluent/ksql/execution/streams/timestamp/MetadataTimestampExtractionPolicy.java b/...java/io/confluent/ksql/execution/streams/timestamp/MetadataTimestampExtractionPolicy.java
@@ -33,8 +33,8 @@ public MetadataTimestampExtractionPolicy(final TimestampExtractor timestampExtra
   }
 
   @Override
-  public TimestampExtractor create(final int columnIndex) {
-    return timestampExtractor;
+  public KsqlTimestampExtractor create(final int columnIndex) {
+    return new MetadataTimestampExtractor(timestampExtractor);
   }
 
   @Override