Kafka 2.1.0 client

benchmarks/src/it/scala/akka/kafka/benchmarks/BenchmarksBase.scala

@@ -12,10 +12,10 @@ abstract class BenchmarksBase(name: String) extends ScalatestKafkaSpec(0) with F
   override def bootstrapServers: String = (1 to BuildInfo.kafkaScale).map(i => sys.props(s"kafka_${i}_9094")).mkString(",")
 
   override def setUp(): Unit = {
+    super.setUp()
     waitUntilCluster() {
       _.nodes().get().size == BuildInfo.kafkaScale
     }
-    super.setUp()
   }
 }
 

build.sbt

@@ -4,8 +4,8 @@ enablePlugins(AutomateHeaderPlugin)
 name := "akka-stream-kafka"
 
 val akkaVersion = "2.5.19"
-val kafkaVersion = "2.0.0"
-val kafkaVersionForDocs = "20"
+val kafkaVersion = "2.1.0"
+val kafkaVersionForDocs = "21"
 val scalatestVersion = "3.0.5"
 val junit4Version = "4.12"
 val slf4jVersion = "1.7.25"

docs/src/main/paradox/home.md

@@ -12,6 +12,7 @@ This **Alpakka Kafka connector** lets you connect [Apache Kafka](https://kafka.a
 
 |Kafka  | Akka version | Alpakka Kafka Connector
 |-------|--------------|-------------------------
+|[2.1.x](https://dist.apache.org/repos/dist/release/kafka/2.1.0/RELEASE_NOTES.html) | 2.5.x        | @ref:[release 1.0-RC1](release-notes/1.0-RC1.md)
 |2.0.x  | 2.5.x        | @ref:[release 1.0-M1](release-notes/1.0-M1.md)
 |1.1.x  | 2.5.x        | [release 0.20+](https://github.com/akka/reactive-kafka/releases)
 |1.0.x  | 2.5.x        | [release 0.20+](https://github.com/akka/reactive-kafka/releases)

testkit/src/main/scala/akka/kafka/testkit/scaladsl/KafkaSpec.scala

@@ -160,7 +160,9 @@ abstract class KafkaSpec(val kafkaPort: Int, val zooKeeperPort: Int, actorSystem
   )(predicate: T => Boolean): Unit = {
     @tailrec def check(triesLeft: Int): Unit =
       Try(predicate(data())).recover {
-        case e: org.apache.kafka.common.errors.TimeoutException => false
+        case ex =>
+          log.debug(s"Ignoring [${ex.getClass.getName}: ${ex.getMessage}] while waiting for desired state")
+          false
       } match {
         case Success(false) if triesLeft > 0 =>
           sleepQuietly(sleepInBetween)

tests/src/test/java/docs/javadsl/FetchMetadataTest.java

@@ -11,7 +11,7 @@
 import akka.kafka.KafkaConsumerActor;
 import akka.kafka.KafkaPorts;
 import akka.kafka.Metadata;
-import akka.pattern.PatternsCS;
+import akka.pattern.Patterns;
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
@@ -73,7 +73,7 @@ public void demo() throws Exception {
     ActorRef consumer = system().actorOf((KafkaConsumerActor.props(settings)));
 
     CompletionStage<Metadata.Topics> topicsStage =
-        PatternsCS.ask(consumer, Metadata.createListTopics(), timeout)
+        Patterns.ask(consumer, Metadata.createListTopics(), timeout)
             .thenApply(reply -> ((Metadata.Topics) reply));
 
     // convert response

tests/src/test/java/docs/javadsl/TransactionsExampleTest.java

@@ -138,7 +138,7 @@ public void usingRestartSource() throws Exception {
                             })
                         .via(Transactional.flow(producerSettings, transactionalId)));
 
-    stream.runWith(Sink.ignore(), materializer);
+    CompletionStage<Done> streamCompletion = stream.runWith(Sink.ignore(), materializer);
 
     // Add shutdown hook to respond to SIGTERM and gracefully shutdown stream
     Runtime.getRuntime().addShutdownHook(new Thread(() -> innerControl.get().shutdown()));
@@ -150,5 +150,6 @@ public void usingRestartSource() throws Exception {
     assertDone(consumer.isShutdown());
     assertDone(innerControl.get().shutdown());
     assertEquals(messages, resultOf(consumer.drainAndShutdown(ec)).size());
+    assertDone(streamCompletion);
   }
 }