Merge branch '6.2.x'

Author: mjsax

README.md

@@ -66,6 +66,7 @@ Additional examples may be found under [src/main/](src/main/java/io/confluent/ex
 | MapFunction                 | DSL, stateless transformations, `map()`                  | [Java 8+ example](src/main/java/io/confluent/examples/streams/MapFunctionLambdaExample.java) | | [Scala Example](src/main/scala/io/confluent/examples/streams/MapFunctionScalaExample.scala) |
 | SessionWindows              | Sessionization of user events, user behavior analysis    | | [Java 7+ example](src/main/java/io/confluent/examples/streams/SessionWindowsExample.java)
 | GlobalKTable                | `join()` between `KStream` and `GlobalKTable`            | [Java 8+ example](src/main/java/io/confluent/examples/streams/GlobalKTablesExample.java) | | |
+| GlobalStore                 | "join" between `KStream` and `GlobalStore`               | [Java 8+ example](src/main/java/io/confluent/examples/streams/GlobalStoresExample.java) | | |
 | PageViewRegion              | `join()` between `KStream` and `KTable`                  | [Java 8+ example](src/main/java/io/confluent/examples/streams/PageViewRegionLambdaExample.java) | [Java 7+ example](src/main/java/io/confluent/examples/streams/PageViewRegionExample.java) | |
 | PageViewRegionGenericAvro   | Working with data in Generic Avro format                 | [Java 8+ example](src/main/java/io/confluent/examples/streams/PageViewRegionLambdaExample.java) | [Java 7+ example](src/main/java/io/confluent/examples/streams/PageViewRegionExample.java) | |
 | WikipediaFeedSpecificAvro   | Working with data in Specific Avro format                | [Java 8+ example](src/main/java/io/confluent/examples/streams/WikipediaFeedAvroLambdaExample.java) | [Java 7+ example](src/main/java/io/confluent/examples/streams/WikipediaFeedAvroExample.java) | |
@@ -109,6 +110,7 @@ Additional examples may be found under [src/test/](src/test/java/io/confluent/ex
 | CustomStreamTableJoin               | DSL, Processor API, Transformers            | [Java 8+ Example](src/test/java/io/confluent/examples/streams/CustomStreamTableJoinIntegrationTest.java) | | |
 | EventDeduplication                  | DSL, Processor API, Transformers            | [Java 8+ Example](src/test/java/io/confluent/examples/streams/EventDeduplicationLambdaIntegrationTest.java) | | |
 | GlobalKTable                        | DSL, global state                           | | [Java 7+ Example](src/test/java/io/confluent/examples/streams/GlobalKTablesExampleTest.java) | |
+| GlobalStore                         | DSL, global state, Transformers             | | [Java 7+ Example](src/test/java/io/confluent/examples/streams/GlobalStoresExampleTest.java) | |
 | HandlingCorruptedInputRecords       | DSL, `flatMap()`                            | [Java 8+ Example](src/test/java/io/confluent/examples/streams/HandlingCorruptedInputRecordsIntegrationTest.java) | | |
 | KafkaMusic (Interactive Queries)    | Interactive Queries, State Stores, REST API | | [Java 7+ Example](src/test/java/io/confluent/examples/streams/interactivequeries/kafkamusic/KafkaMusicExampleTest.java) | |
 | MapFunction                         | DSL, stateless transformations, `map()`     | [Java 8+ Example](src/test/java/io/confluent/examples/streams/MapFunctionLambdaIntegrationTest.java) | | |

src/main/java/io/confluent/examples/streams/AnomalyDetectionLambdaExample.java

@@ -160,7 +160,7 @@ public static void main(final String[] args) {
     // Always (and unconditionally) clean local state prior to starting the processing topology.
     // We opt for this unconditional call here because this will make it easier for you to play around with the example
     // when resetting the application for doing a re-run (via the Application Reset Tool,
-    // http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool).
+    // https://docs.confluent.io/platform/current/streams/developer-guide/app-reset-tool.html).
     //
     // The drawback of cleaning up local state prior is that your app must rebuilt its local state from scratch, which
     // will take time and will require reading all the state-relevant data from the Kafka cluster over the network.

src/main/java/io/confluent/examples/streams/ApplicationResetExample.java

@@ -28,7 +28,7 @@
 
 /**
  * Demonstrates how to reset a Kafka Streams application to re-process its input data from scratch.
- * See also <a href='http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool'>http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool</a>
+ * See also <a href='https://docs.confluent.io/platform/current/streams/developer-guide/app-reset-tool.html'>https://docs.confluent.io/platform/current/streams/developer-guide/app-reset-tool.html</a>
  * <p>
  * The main purpose of the example is to explain the usage of the "Application Reset Tool".
  * Thus, we don’t put the focus on what this topology is actually doing&mdash;the point is to have an example of a

src/main/java/io/confluent/examples/streams/GlobalKTablesExampleDriver.java renamed to src/main/java/io/confluent/examples/streams/GlobalKTablesAndStoresExampleDriver.java

@@ -19,7 +19,7 @@
 import io.confluent.examples.streams.avro.EnrichedOrder;
 import io.confluent.examples.streams.avro.Order;
 import io.confluent.examples.streams.avro.Product;
-import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
+import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
 import io.confluent.kafka.serializers.KafkaAvroDeserializer;
 import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;
 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
@@ -46,8 +46,8 @@
 import static io.confluent.examples.streams.GlobalKTablesExample.PRODUCT_TOPIC;
 
 /**
- * This is a sample driver for the {@link GlobalKTablesExample}.
- * To run this driver please first refer to the instructions in {@link GlobalKTablesExample}.
+ * This is a sample driver for the {@link GlobalKTablesExample} and {@link GlobalStoresExample}.
+ * To run this driver please first refer to the instructions in {@link GlobalKTablesExample} or {@link GlobalStoresExample}.
  * You can then run this class directly in your IDE or via the command line.
  * <p>
  * To run via the command line you might want to package as a fatjar first. Please refer to:
@@ -56,11 +56,11 @@
  * Once packaged you can then run:
  * <pre>
  * {@code
- * $ java -cp target/kafka-streams-examples-7.0.0-0-standalone.jar io.confluent.examples.streams.GlobalKTablesExampleDriver
+ * $ java -cp target/kafka-streams-examples-7.0.0-0-standalone.jar io.confluent.examples.streams.GlobalKTablesAndStoresExampleDriver
  * }
  * </pre>
  */
-public class GlobalKTablesExampleDriver {
+public class GlobalKTablesAndStoresExampleDriver {
 
   private static final Random RANDOM = new Random();
   private static final int RECORDS_TO_GENERATE = 100;
@@ -85,7 +85,7 @@ private static void receiveEnrichedOrders(
     consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
     consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, Serdes.Long().deserializer().getClass());
     consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
-    consumerProps.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
+    consumerProps.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
     consumerProps.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
 
     final KafkaConsumer<Long, EnrichedOrder> consumer = new KafkaConsumer<>(consumerProps);
@@ -184,7 +184,7 @@ private static <VT extends SpecificRecord> SpecificAvroSerde<VT> createSerde(fin
 
     final SpecificAvroSerde<VT> serde = new SpecificAvroSerde<>();
     final Map<String, String> serdeConfig = Collections.singletonMap(
-        AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
+        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
     serde.configure(serdeConfig, false);
     return serde;
   }

src/main/java/io/confluent/examples/streams/GlobalKTablesExample.java

@@ -42,6 +42,10 @@
  * Demonstrates how to perform joins between  KStreams and GlobalKTables, i.e. joins that
  * don't require re-partitioning of the input streams.
  * <p>
+ * The {@link GlobalStoresExample} shows another way to perform the same operation using
+ * {@link org.apache.kafka.streams.TopologyDescription.GlobalStore} and a
+ * {@link org.apache.kafka.streams.kstream.ValueTransformer}.
+ * <p>
  * In this example, we join a stream of orders that reads from a topic named
  * "order" with a customers table that reads from a topic named "customer", and a products
  * table that reads from a topic "product". The join produces an EnrichedOrder object.
@@ -75,14 +79,14 @@
  * $ java -cp target/kafka-streams-examples-7.0.0-0-standalone.jar io.confluent.examples.streams.GlobalKTablesExample
  * }
  * </pre>
- * 4) Write some input data to the source topics (e.g. via {@link GlobalKTablesExampleDriver}). The
+ * 4) Write some input data to the source topics (e.g. via {@link GlobalKTablesAndStoresExampleDriver}). The
  * already running example application (step 3) will automatically process this input data and write
  * the results to the output topic.
  * <pre>
  * {@code
  * # Here: Write input data using the example driver. The driver will exit once it has received
  * # all EnrichedOrders
- * $ java -cp target/kafka-streams-examples-7.0.0-0-standalone.jar io.confluent.examples.streams.GlobalKTablesExampleDriver
+ * $ java -cp target/kafka-streams-examples-7.0.0-0-standalone.jar io.confluent.examples.streams.GlobalKTablesAndStoresExampleDriver
  * }
  * </pre>
  * <p>
@@ -108,7 +112,7 @@ public static void main(final String[] args) {
     // Always (and unconditionally) clean local state prior to starting the processing topology.
     // We opt for this unconditional call here because this will make it easier for you to play around with the example
     // when resetting the application for doing a re-run (via the Application Reset Tool,
-    // http://docs.confluent.io/current/streams/developer-guide.html#application-reset-tool).
+    // https://docs.confluent.io/platform/current/streams/developer-guide/app-reset-tool.html).
     //
     // The drawback of cleaning up local state prior is that your app must rebuilt its local state from scratch, which
     // will take time and will require reading all the state-relevant data from the Kafka cluster over the network.