Upgrade Kafka 3.0.0

Author: perkss

.github/workflows/build.yml

@@ -0,0 +1,12 @@
+name: Java CI
+
+on: [ push ]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v2
+      - name: Build with Maven
+        run: mvn install -Dmaven.compiler.target=11 -Dmaven.compiler.source=11

.travis.yml

@@ -1,36 +1,48 @@
 # Reactive Kotlin Kafka and Kafka Streams with Kotlin
+
 Kafka and Kafka Stream Examples in Kotlin with Project Reactor Kafka
 
 Please check out the blog posts for more details.
-* [Kotlin Kafka Streams](https://perkss.github.io/#/DistributedSystems/Streaming#KafkaStreamsKotlin) 
+
+* [Kotlin Kafka Streams](https://perkss.github.io/#/DistributedSystems/Streaming#KafkaStreamsKotlin)
 
 ## Docker Environment
-Please use the docker-compose file in the root of each module project to create the Kafka Brokers and Zookeeper and where appropriate 
-the Schema Registry.
+
+Please use the docker-compose file in the root of each module project to create the Kafka Brokers and Zookeeper and
+where appropriate the Schema Registry.
 
 Please check the directory README for details how to run this example.
 
 ## Integration Tests
-Integration tests can be found for each module project and these require Docker to be running and use [Testcontainers](https://www.testcontainers.org/) these are powerful tests that fire up Kafka instances and our application and fully test the flow of messages through our streaming application.
+
+Integration tests can be found for each module project and these require Docker to be running and
+use [Testcontainers](https://www.testcontainers.org/) these are powerful tests that fire up Kafka instances and our
+application and fully test the flow of messages through our streaming application.
 
 ## Kafka Reactive Producer Consumer
-This example shows how you can use the reactive API to build a consumer from a `lowercase-topic` map the data and output it
-with the same key to a `uppercase-topic` with the data converted to uppercase. Please check the sub module README for 
-how to execute this. Its a very interesting yet simple example, as you can see when the consume is lazily instantiated when 
-it connects and then once a message is received it lazily instantiates the producer to send on.
+
+This example shows how you can use the reactive API to build a consumer from a `lowercase-topic` map the data and output
+it with the same key to a `uppercase-topic` with the data converted to uppercase. Please check the sub module README for
+how to execute this. It is a very interesting yet simple example, as you can see when the consume is lazily instantiated
+when it connects and then once a message is received it lazily instantiates the producer to send on.
 
 ## Kafka Reactive Secure Producer Consumer
-Shows how you can run a secured broker cluster using TLS and a application that will will consume and produce with this secure 
-transport layer to the brokers. Details can be found in the sub folder README.
+
+Shows how you can run a secured broker cluster using TLS and a application that will will consume and produce with this
+secure transport layer to the brokers. Details can be found in the sub folder README.
 
 ## Kafka Streams and Kotlin Examples
-This module is for examples of using Kafka Streams with Kotlin and Avro. Here we build a stock ordering system that has the concept
- of customers to place orders. We use Avro to define schemas for the main topics and use changelog tables to store down product and 
- customer information which is joined to the OrderRequests. This module depends on the Avro code generation in the `avro-schemas` module
- so that needs building before compiling this module.
- 
- ### Kafka Streams Examples
- We also have a package for examples to show the following features [found here.](kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/examples)
- * Windowing
- * Aggregates
- * Joins
+
+This module is for examples of using Kafka Streams with Kotlin and Avro. Here we build a stock ordering system that has
+the concept of customers to place orders. We use Avro to define schemas for the main topics and use changelog tables to
+store down product and customer information which is joined to the OrderRequests. This module depends on the Avro code
+generation in the `avro-schemas` module so that needs building before compiling this module.
+
+### Kafka Streams Examples
+
+We also have a package for examples to show the following
+features [found here.](kotlin-kafka-streams-examples/src/main/kotlin/com/perkss/kafka/reactive/examples)
+
+* Windowing
+* Aggregates
+* Joins

README.md

@@ -1,6 +1,6 @@
 # Avro Schemas
 
-This module is used to show the automatic Class creation using the `avro-maven-plugin` from a `avsc` file.
-Here we create the order request data type to be used by the `SpecificAvroSerde`.
+This module is used to show the automatic Class creation using the `avro-maven-plugin` from a `avsc` file. Here we
+create the order request data type to be used by the `SpecificAvroSerde`.
 
 `kotlin-kafka-streams` depends on this module.

avro-schemas/README.md

@@ -1,9 +1,10 @@
 # Kafka Reactive Producer Consumer Example
 
 ## Check the TestContainers Test
-To quickly run the application without any setup you just need a running Docker process. This test will fire up a Kafka Container,
-and then the topology as a spring boot application, push a message on the `lowercase-topic`, process it
-in the topology to uppercase and output it to the `uppercase-topic`
+
+To quickly run the application without any setup you just need a running Docker process. This test will fire up a Kafka
+Container, and then the topology as a spring boot application, push a message on the `lowercase-topic`, process it in
+the topology to uppercase and output it to the `uppercase-topic`
 
 ## Running the Example
 
@@ -27,12 +28,13 @@ Check the topic was created:
 docker exec reactive-kafka-example-broker kafka-topics --zookeeper zookeeper:2181 --list
 ```
 
-Produce some example data to the topic and with the Spring Boot Kafka Reactive App Running it will log the consumption of these messages:
+Produce some example data to the topic and with the Spring Boot Kafka Reactive App Running it will log the consumption
+of these messages:
 
 ```bash
 docker exec -it reactive-kafka-example-broker kafka-console-producer --broker-list broker:9092 --topic lowercase-topic --property "parse.key=true" --property "key.separator=:"
 ```
-              
+
 For example
 
 ```
@@ -45,16 +47,18 @@ Check the topology output is uppercase with the console consumer
 docker exec reactive-kafka-example-broker kafka-console-consumer --bootstrap-server broker:9092 --topic uppercase-topic --property print.key=true --property key.separator="-" --from-beginning
 ```
 
-
 ## Bad message fixes just like prod support
+
 If you input a bad message onto a topic you can move the offset to the latest
+
 ```bash
 docker exec -it reactive-kafka-example-broker kafka-consumer-groups --bootstrap-server broker:9092 --group sample-group --reset-offsets --to-latest --topic lowercase-topic --execute
 ```
 
-
 ## Shell into the Container
-If you need to shell into containers you can do like this. 
+
+If you need to shell into containers you can do like this.
+
 ```bash
 sudo docker exec -i -t reactive-kafka-example-zookeeper /bin/bash
 ```

kafka-reactive-producer-consumer/README.md

@@ -63,6 +63,7 @@
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>testcontainers</artifactId>
+            <version>${testcontainers.version}</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -73,7 +74,6 @@
         <dependency>
             <groupId>org.testcontainers</groupId>
             <artifactId>kafka</artifactId>
-            <version>1.12.1</version>
             <scope>test</scope>
         </dependency>
         <dependency>
@@ -85,7 +85,7 @@
         <dependency>
             <groupId>org.awaitility</groupId>
             <artifactId>awaitility-kotlin</artifactId>
-            <version>4.0.2</version>
+            <version>4.1.1</version>
             <scope>test</scope>
         </dependency>
     </dependencies>

kafka-reactive-producer-consumer/pom.xml

@@ -10,9 +10,11 @@ import reactor.core.publisher.Mono
 import reactor.kafka.sender.SenderRecord
 
 @SpringBootApplication
-class KafkaReactiveApp(private var consumer: KafkaReactiveConsumer<String, String>,
-                       private var producer: KafkaReactiveProducer<String, String>,
-                       private var reactiveKafkaAppProperties: ReactiveKafkaAppProperties) : CommandLineRunner {
+class KafkaReactiveApp(
+    private var consumer: KafkaReactiveConsumer<String, String>,
+    private var producer: KafkaReactiveProducer<String, String>,
+    private var reactiveKafkaAppProperties: ReactiveKafkaAppProperties
+) : CommandLineRunner {
 
     companion object {
         private val logger = LoggerFactory.getLogger(KafkaReactiveApp::class.java)
@@ -25,16 +27,16 @@ class KafkaReactiveApp(private var consumer: KafkaReactiveConsumer<String, Strin
         val outputTopic = reactiveKafkaAppProperties.outputTopic
 
         consumer.consume()
-                .map {
-                    val producerRecord = ProducerRecord(outputTopic, it.key(), it.value().toUpperCase())
-                    logger.info("Building uppercase message. Key: ${it.key()} Message: ${it.value().toUpperCase()}")
-                    SenderRecord.create(producerRecord, it.key())
-                }
-                .map { producer.send(Mono.just(it)).subscribe() }
-                .doOnError { logger.error("An error has occurred $it") }
-                .subscribe {
-                    logger.info("Subscribing to Consumer and Producer")
-                }
+            .map {
+                val producerRecord = ProducerRecord(outputTopic, it.key(), it.value().toUpperCase())
+                logger.info("Building uppercase message. Key: ${it.key()} Message: ${it.value().toUpperCase()}")
+                SenderRecord.create(producerRecord, it.key())
+            }
+            .map { producer.send(Mono.just(it)).subscribe() }
+            .doOnError { logger.error("An error has occurred $it") }
+            .subscribe {
+                logger.info("Subscribing to Consumer and Producer")
+            }
     }
 
 }

kafka-reactive-producer-consumer/src/main/kotlin/com/perkss/kafka/reactive/KafkaReactiveApp.kt

@@ -9,10 +9,12 @@ import reactor.kafka.receiver.ReceiverOptions
 import reactor.kafka.receiver.ReceiverRecord
 import java.util.*
 
-class KafkaReactiveConsumer<K, V>(bootstrapServers: String,
-                                  topic: String,
-                                  consumerGroupId: String,
-                                  autoOffsetReset: String = "earliest") {
+class KafkaReactiveConsumer<K, V>(
+    bootstrapServers: String,
+    topic: String,
+    consumerGroupId: String,
+    autoOffsetReset: String = "earliest"
+) {
 
     companion object {
         private val logger = LoggerFactory.getLogger(KafkaReactiveConsumer::class.java)
@@ -30,12 +32,12 @@ class KafkaReactiveConsumer<K, V>(bootstrapServers: String,
         val consumerOptions = ReceiverOptions.create<K, V>(consumerProps).subscription(Collections.singleton(topic))
 
         receiver = KafkaReceiver.create(consumerOptions)
-                .receive()
-                .map {
-                    logger.info("Received message: $it")
-                    it.receiverOffset().commit()
-                    it
-                }
+            .receive()
+            .map {
+                logger.info("Received message: $it")
+                it.receiverOffset().commit()
+                it
+            }
     }
 
     fun consume() = receiver

kafka-reactive-producer-consumer/src/main/kotlin/com/perkss/kafka/reactive/KafkaReactiveConsumer.kt

@@ -31,7 +31,13 @@ class KafkaReactiveProducer<K, V>(bootstrapServers: String) {
 
     fun send(outboundFlux: Mono<SenderRecord<K, V, String>>): Flux<SenderResult<String>> {
         return sender.send(outboundFlux)
-                .doOnError { e -> logger.error("Send failed", e) }
-                .doOnNext { r -> logger.info("Message Key ${r.correlationMetadata()} send response: ${r.recordMetadata().topic()}") }
+            .doOnError { e -> logger.error("Send failed", e) }
+            .doOnNext { r ->
+                logger.info(
+                    "Message Key ${r.correlationMetadata()} send response: ${
+                        r.recordMetadata().topic()
+                    }"
+                )
+            }
     }
 }

kafka-reactive-producer-consumer/src/main/kotlin/com/perkss/kafka/reactive/KafkaReactiveProducer.kt

@@ -13,14 +13,15 @@ class AppConfig {
 
     @Bean
     fun reactiveProducer(propertiesReactiveKafka: ReactiveKafkaAppProperties) =
-            KafkaReactiveProducer<String, String>(propertiesReactiveKafka.bootstrapServers)
+        KafkaReactiveProducer<String, String>(propertiesReactiveKafka.bootstrapServers)
 
     @Bean
     fun reactiveConsumer(propertiesReactiveKafka: ReactiveKafkaAppProperties) =
-            KafkaReactiveConsumer<String, String>(
-                    propertiesReactiveKafka.bootstrapServers,
-                    propertiesReactiveKafka.inputTopic,
-                    propertiesReactiveKafka.consumerGroupId)
+        KafkaReactiveConsumer<String, String>(
+            propertiesReactiveKafka.bootstrapServers,
+            propertiesReactiveKafka.inputTopic,
+            propertiesReactiveKafka.consumerGroupId
+        )
 
 
 }