Use streams 1.0 api methods (confluentinc#365)

ksql-engine/src/main/java/io/confluent/ksql/QueryEngine.java

@@ -71,8 +71,8 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.streams.KafkaStreams;
+import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -212,7 +212,7 @@ public void buildQueryPhysicalPlan(final List<QueryMetadata> physicalPlans,
                                      final boolean updateMetastore) throws Exception {
 
     PlanNode logicalPlan = statementPlanPair.getRight();
-    KStreamBuilder builder = new KStreamBuilder();
+    StreamsBuilder builder = new StreamsBuilder();
 
     KsqlConfig ksqlConfigClone = ksqlEngine.getKsqlConfig().clone();
 
@@ -272,7 +272,7 @@ public void buildQueryPhysicalPlan(final List<QueryMetadata> physicalPlans,
    */
   private QueryMetadata buildPlanForBareQuery(boolean addUniqueTimeSuffix,
                                               Pair<String, PlanNode> statementPlanPair, Map<String, Object> overriddenStreamsProperties,
-                                              KStreamBuilder builder, KsqlConfig ksqlConfigClone, QueuedSchemaKStream schemaKStream,
+                                              StreamsBuilder builder, KsqlConfig ksqlConfigClone, QueuedSchemaKStream schemaKStream,
                                               KsqlBareOutputNode bareOutputNode, String serviceId, String transientQueryPrefix) {
 
     String applicationId = getBareQueryApplicationId(serviceId, transientQueryPrefix);
@@ -309,7 +309,7 @@ private QueryMetadata buildPlanForBareQuery(boolean addUniqueTimeSuffix,
    */
   private QueryMetadata buildPlanForStructuredOutputNode(boolean addUniqueTimeSuffix,
                                                          Pair<String, PlanNode> statementPlanPair, Map<String, Object> overriddenStreamsProperties,
-                                                         boolean updateMetastore, KStreamBuilder builder, KsqlConfig ksqlConfigClone, SchemaKStream schemaKStream,
+                                                         boolean updateMetastore, StreamsBuilder builder, KsqlConfig ksqlConfigClone, SchemaKStream schemaKStream,
                                                          KsqlStructuredDataOutputNode outputNode, String serviceId, String persistanceQueryPrefix) {
 
     long queryId = getNextQueryId();
@@ -408,7 +408,7 @@ public StructuredDataSource getResultDatasource(final Select select, final Strin
   }
 
   private KafkaStreams buildStreams(
-      final KStreamBuilder builder,
+      final StreamsBuilder builder,
       final String applicationId,
       final KsqlConfig ksqlConfig,
       final Map<String, Object> overriddenProperties
@@ -430,11 +430,11 @@ private KafkaStreams buildStreams(
           KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX,
           ksqlConfig.get(KsqlConfig.KSQL_TIMESTAMP_COLUMN_INDEX));
       newStreamsProperties.put(
-          StreamsConfig.TIMESTAMP_EXTRACTOR_CLASS_CONFIG, KsqlTimestampExtractor.class);
+          StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG, KsqlTimestampExtractor.class);
     }
 
 
-    return new KafkaStreams(builder, new StreamsConfig(newStreamsProperties));
+    return new KafkaStreams(builder.build(), new StreamsConfig(newStreamsProperties));
   }
 
   private long getNextQueryId() {

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java

@@ -56,14 +56,16 @@
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.streams.Consumed;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
+import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.kstream.Windowed;
-import org.apache.kafka.streams.processor.TopologyBuilder;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -76,7 +78,7 @@
 
 public class PhysicalPlanBuilder {
 
-  private final KStreamBuilder builder;
+  private final StreamsBuilder builder;
   private final KsqlConfig ksqlConfig;
   private final KafkaTopicClient kafkaTopicClient;
   private final MetastoreUtil metastoreUtil;
@@ -102,7 +104,7 @@ public class PhysicalPlanBuilder {
 
   private OutputNode planSink = null;
 
-  public PhysicalPlanBuilder(final KStreamBuilder builder,
+  public PhysicalPlanBuilder(final StreamsBuilder builder,
                              final KsqlConfig ksqlConfig,
                              final KafkaTopicClient kafkaTopicClient,
                              final MetastoreUtil metastoreUtil) {
@@ -425,8 +427,8 @@ private SchemaKStream buildSource(final SourceNode sourceNode, Map<String, Objec
 
       return new SchemaKStream(sourceNode.getSchema(),
           builder
-              .stream((TopologyBuilder.AutoOffsetReset) null, Serdes.String(), genericRowSerde,
-                  structuredDataSourceNode.getStructuredDataSource().getKsqlTopic().getKafkaTopicName())
+              .stream(structuredDataSourceNode.getStructuredDataSource().getKsqlTopic().getKafkaTopicName(),
+                  Consumed.with(Serdes.String(), genericRowSerde))
               .map(nonWindowedMapper)
               .transformValues(new AddTimestampColumnValueTransformerSupplier()),
           sourceNode.getKeyField(), new ArrayList<>(),
@@ -435,44 +437,45 @@ private SchemaKStream buildSource(final SourceNode sourceNode, Map<String, Objec
     throw new KsqlException("Unsupported source logical node: " + sourceNode.getClass().getName());
   }
 
-  private <K> KTable table(final KStream<K, GenericRow> stream, final Serde<K> keySerde, final Serde<GenericRow> valueSerde, final String stateStoreName) {
-    return stream.groupByKey(keySerde, valueSerde)
-        .reduce((genericRow, newValue) -> newValue, stateStoreName);
+  private <K> KTable table(final KStream<K, GenericRow> stream, final Serde<K> keySerde, final Serde<GenericRow> valueSerde) {
+    return stream.groupByKey(Serialized.with(keySerde, valueSerde))
+        .reduce((genericRow, newValue) -> newValue);
   }
 
   @SuppressWarnings("unchecked")
-  private KTable createKTable(final TopologyBuilder.AutoOffsetReset autoOffsetReset,
+  private KTable createKTable(final Topology.AutoOffsetReset autoOffsetReset,
                               final KsqlTable ksqlTable,
                               final Serde<GenericRow> genericRowSerde,
                               final Serde<GenericRow> genericRowSerdeAfterRead) {
     if (ksqlTable.isWindowed()) {
       return table(builder
-          .stream(autoOffsetReset, windowedSerde, genericRowSerde,
-              ksqlTable.getKsqlTopic().getKafkaTopicName())
+          .stream(ksqlTable.getKsqlTopic().getKafkaTopicName(),
+              Consumed.with(windowedSerde, genericRowSerde)
+                  .withOffsetResetPolicy(autoOffsetReset))
           .map(windowedMapper)
-          .transformValues(new AddTimestampColumnValueTransformerSupplier()), windowedSerde, genericRowSerdeAfterRead, ksqlTable.getStateStoreName());
+          .transformValues(new AddTimestampColumnValueTransformerSupplier()), windowedSerde, genericRowSerdeAfterRead);
     } else {
       return table(builder
-              .stream(autoOffsetReset, Serdes.String(), genericRowSerde,
-                  ksqlTable.getKsqlTopic().getKafkaTopicName())
+              .stream(ksqlTable.getKsqlTopic().getKafkaTopicName(),
+                  Consumed.with(Serdes.String(), genericRowSerde)
+                      .withOffsetResetPolicy(autoOffsetReset))
               .map(nonWindowedMapper)
               .transformValues(new AddTimestampColumnValueTransformerSupplier()),
-          Serdes.String(), genericRowSerdeAfterRead, ksqlTable.getStateStoreName());
+          Serdes.String(), genericRowSerdeAfterRead);
     }
   }
 
-  private TopologyBuilder.AutoOffsetReset getAutoOffsetReset(Map<String, Object> props) {
-    TopologyBuilder.AutoOffsetReset autoOffsetReset = null;
+  private Topology.AutoOffsetReset getAutoOffsetReset(Map<String, Object> props) {
     if (props.containsKey(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG)) {
       if (props.get(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG).toString()
           .equalsIgnoreCase("EARLIEST")) {
-        autoOffsetReset = TopologyBuilder.AutoOffsetReset.EARLIEST;
+        return Topology.AutoOffsetReset.EARLIEST;
       } else if (props.get(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG).toString()
           .equalsIgnoreCase("LATEST")) {
-        autoOffsetReset = TopologyBuilder.AutoOffsetReset.LATEST;
+        return Topology.AutoOffsetReset.LATEST;
       }
     }
-    return autoOffsetReset;
+    return null;
   }
 
   private SchemaKStream buildJoin(final JoinNode joinNode, final Map<String, Object> propsMap)
@@ -529,7 +532,7 @@ private KsqlTopicSerDe getResultTopicSerde(final PlanNode node) {
   }
 
 
-  public KStreamBuilder getBuilder() {
+  public StreamsBuilder getBuilder() {
     return builder;
   }
 

ksql-engine/src/main/java/io/confluent/ksql/planner/plan/KsqlStructuredDataOutputNode.java

@@ -20,7 +20,6 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import io.confluent.ksql.metastore.KsqlTopic;
 import io.confluent.ksql.serde.KsqlTopicSerDe;
-import io.confluent.ksql.util.KsqlConfig;
 
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKGroupedStream.java

@@ -24,14 +24,18 @@
 import io.confluent.ksql.GenericRow;
 import io.confluent.ksql.util.KsqlException;
 import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.streams.kstream.Initializer;
 import org.apache.kafka.streams.kstream.KGroupedStream;
 import org.apache.kafka.streams.kstream.KTable;
+import org.apache.kafka.streams.kstream.Materialized;
 import org.apache.kafka.streams.kstream.SessionWindows;
 import org.apache.kafka.streams.kstream.TimeWindows;
 import org.apache.kafka.streams.kstream.Windowed;
+import org.apache.kafka.streams.state.SessionStore;
+import org.apache.kafka.streams.state.WindowStore;
 
 import java.util.List;
 
@@ -51,6 +55,7 @@ public class SchemaKGroupedStream {
     this.sourceSchemaKStreams = sourceSchemaKStreams;
   }
 
+  @SuppressWarnings("unchecked")
   public SchemaKTable aggregate(final Initializer initializer,
                                 final KudafAggregator aggregator,
                                 final WindowExpression windowExpression,
@@ -60,42 +65,38 @@ public SchemaKTable aggregate(final Initializer initializer,
     KTable<Windowed<String>, GenericRow> aggKtable;
     if (windowExpression != null) {
       isWindowed = true;
+      final Materialized<String, GenericRow, WindowStore<Bytes, byte[]>> materialized
+          = Materialized.<String, GenericRow, WindowStore<Bytes, byte[]>>as(storeName)
+          .withValueSerde(topicValueSerDe);
       if (windowExpression.getKsqlWindowExpression() instanceof TumblingWindowExpression) {
         TumblingWindowExpression tumblingWindowExpression =
             (TumblingWindowExpression) windowExpression.getKsqlWindowExpression();
         aggKtable =
-            kgroupedStream
+            kgroupedStream.windowedBy(TimeWindows.of(tumblingWindowExpression.getSizeUnit().toMillis(tumblingWindowExpression.getSize())))
                 .aggregate(initializer, aggregator,
-                                     TimeWindows.of(tumblingWindowExpression.getSizeUnit().toMillis(tumblingWindowExpression.getSize())),
-                           topicValueSerDe,
-                           storeName);
+                    materialized);
       } else if (windowExpression.getKsqlWindowExpression() instanceof HoppingWindowExpression) {
         HoppingWindowExpression hoppingWindowExpression =
             (HoppingWindowExpression) windowExpression.getKsqlWindowExpression();
         aggKtable =
-            kgroupedStream
-                .aggregate(initializer, aggregator,
-                           TimeWindows.of(
-                               hoppingWindowExpression.getSizeUnit().toMillis(hoppingWindowExpression.getSize()))
-                                         .advanceBy(
-                                             hoppingWindowExpression.getAdvanceByUnit().toMillis(hoppingWindowExpression.getAdvanceBy())),
-                                     topicValueSerDe, storeName);
+            kgroupedStream.windowedBy(TimeWindows.of(
+                hoppingWindowExpression.getSizeUnit().toMillis(hoppingWindowExpression.getSize()))
+                .advanceBy(
+                    hoppingWindowExpression.getAdvanceByUnit().toMillis(hoppingWindowExpression.getAdvanceBy())))
+                .aggregate(initializer, aggregator, materialized);
       } else if (windowExpression.getKsqlWindowExpression() instanceof SessionWindowExpression) {
         SessionWindowExpression sessionWindowExpression =
             (SessionWindowExpression) windowExpression.getKsqlWindowExpression();
         aggKtable =
-            kgroupedStream
-                .aggregate(initializer, aggregator,
-                           aggregator.getMerger(),
-                           SessionWindows.with(sessionWindowExpression.getSizeUnit().toMillis(sessionWindowExpression.getGap())),
-                           topicValueSerDe,
-                           storeName);
+            kgroupedStream.windowedBy(SessionWindows.with(sessionWindowExpression.getSizeUnit().toMillis(sessionWindowExpression.getGap())))
+                .aggregate(initializer, aggregator, aggregator.getMerger(),
+                           Materialized.<String, GenericRow, SessionStore<Bytes, byte[]>>as(storeName).withValueSerde(topicValueSerDe));
       } else {
         throw new KsqlException("Could not set the window expression for aggregate.");
       }
     } else {
       aggKtable =
-          kgroupedStream.aggregate(initializer, aggregator, topicValueSerDe, storeName);
+          kgroupedStream.aggregate(initializer, aggregator, Materialized.with(null, topicValueSerDe));
     }
     return new SchemaKTable(schema, aggKtable, keyField, sourceSchemaKStreams, isWindowed,
                             SchemaKStream.Type.AGGREGATE);

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKStream.java

@@ -36,9 +36,12 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.SchemaBuilder;
 import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.kstream.Joined;
 import org.apache.kafka.streams.kstream.KGroupedStream;
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Produced;
+import org.apache.kafka.streams.kstream.Serialized;
 import org.apache.kafka.streams.kstream.ValueJoiner;
 import org.apache.kafka.streams.kstream.ValueMapper;
 import org.slf4j.Logger;
@@ -94,7 +97,7 @@ public SchemaKStream into(final String kafkaTopicName, final Serde<GenericRow> t
             }
           }
           return new KeyValue<>(key, new GenericRow(columns));
-        }).to(Serdes.String(), topicValueSerDe, kafkaTopicName);
+        }).to(kafkaTopicName, Produced.with(Serdes.String(), topicValueSerDe));
     return this;
   }
 
@@ -200,7 +203,7 @@ public SchemaKStream leftJoin(final SchemaKTable schemaKTable, final Schema join
 
               GenericRow joinGenericRow = new GenericRow(columns);
               return joinGenericRow;
-            }, Serdes.String(), SerDeUtil.getRowSerDe(joinSerDe, this.getSchema()));
+            }, Joined.with(Serdes.String(), SerDeUtil.getRowSerDe(joinSerDe, this.getSchema()), null));
 
     return new SchemaKStream(joinSchema, joinedKStream, joinKey,
                              Arrays.asList(this, schemaKTable), Type.JOIN);
@@ -230,7 +233,7 @@ public SchemaKStream selectKey(final Field newKeyField) {
 
   public SchemaKGroupedStream groupByKey(final Serde keySerde,
                                          final Serde valSerde) {
-    KGroupedStream kgroupedStream = kstream.groupByKey(keySerde, valSerde);
+    KGroupedStream kgroupedStream = kstream.groupByKey(Serialized.with(keySerde, valSerde));
     return new SchemaKGroupedStream(schema, kgroupedStream, keyField, Arrays.asList(this));
   }
 

ksql-engine/src/main/java/io/confluent/ksql/structured/SchemaKTable.java

@@ -35,6 +35,7 @@
 import org.apache.kafka.streams.kstream.KStream;
 import org.apache.kafka.streams.kstream.KTable;
 import org.apache.kafka.streams.kstream.KeyValueMapper;
+import org.apache.kafka.streams.kstream.Produced;
 import org.apache.kafka.streams.kstream.ValueMapper;
 import org.apache.kafka.streams.kstream.Windowed;
 import org.slf4j.Logger;
@@ -80,7 +81,7 @@ public SchemaKTable into(final String kafkaTopicName, final Serde<GenericRow> to
               }
             }
             return new KeyValue<>(key, new GenericRow(columns));
-          }).to(new WindowedSerde(), topicValueSerDe, kafkaTopicName);
+          }).to(kafkaTopicName, Produced.with(new WindowedSerde(), topicValueSerDe));
     } else {
       ktable.toStream()
           .map((KeyValueMapper<String, GenericRow, KeyValue<String, GenericRow>>) (key, row) -> {
@@ -94,7 +95,7 @@ public SchemaKTable into(final String kafkaTopicName, final Serde<GenericRow> to
               }
             }
             return new KeyValue<>(key, new GenericRow(columns));
-          }).to(Serdes.String(), topicValueSerDe, kafkaTopicName);
+          }).to(kafkaTopicName, Produced.with(Serdes.String(), topicValueSerDe));
     }
 
     return this;

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java

@@ -37,7 +37,7 @@
 import io.confluent.ksql.util.MetaStoreFixture;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.streams.kstream.KStreamBuilder;
+import org.apache.kafka.streams.StreamsBuilder;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -48,14 +48,14 @@
 
 public class PhysicalPlanBuilderTest {
 
-    KStreamBuilder kStreamBuilder;
+    StreamsBuilder streamsBuilder;
     KsqlParser ksqlParser;
     PhysicalPlanBuilder physicalPlanBuilder;
     MetaStore metaStore;
 
     @Before
     public void before() {
-        kStreamBuilder = new KStreamBuilder();
+        streamsBuilder = new StreamsBuilder();
         ksqlParser = new KsqlParser();
         metaStore = MetaStoreFixture.getNewMetaStore();
         Map<String, Object> configMap = new HashMap<>();
@@ -64,7 +64,7 @@ public void before() {
         configMap.put("commit.interval.ms", 0);
         configMap.put("cache.max.bytes.buffering", 0);
         configMap.put("auto.offset.reset", "earliest");
-        physicalPlanBuilder = new PhysicalPlanBuilder(kStreamBuilder, new KsqlConfig(configMap), new FakeKafkaTopicClient(), new MetastoreUtil());
+        physicalPlanBuilder = new PhysicalPlanBuilder(streamsBuilder, new KsqlConfig(configMap), new FakeKafkaTopicClient(), new MetastoreUtil());
     }
 
     private SchemaKStream buildPhysicalPlan(String queryStr) throws Exception {