KAFKA-9447: Add new customized EOS model example

clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java

@@ -51,21 +51,18 @@ public Map<String, KafkaFuture<TopicDescription>> values() {
      */
     public KafkaFuture<Map<String, TopicDescription>> all() {
         return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])).
-            thenApply(new KafkaFuture.BaseFunction<Void, Map<String, TopicDescription>>() {
-                @Override
-                public Map<String, TopicDescription> apply(Void v) {
-                    Map<String, TopicDescription> descriptions = new HashMap<>(futures.size());
-                    for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : futures.entrySet()) {
-                        try {
-                            descriptions.put(entry.getKey(), entry.getValue().get());
-                        } catch (InterruptedException | ExecutionException e) {
-                            // This should be unreachable, because allOf ensured that all the futures
-                            // completed successfully.
-                            throw new RuntimeException(e);
-                        }
+            thenApply(v -> {
+                Map<String, TopicDescription> descriptions = new HashMap<>(futures.size());
+                for (Map.Entry<String, KafkaFuture<TopicDescription>> entry : futures.entrySet()) {
+                    try {
+                        descriptions.put(entry.getKey(), entry.getValue().get());
+                    } catch (InterruptedException | ExecutionException e) {
+                        // This should be unreachable, because allOf ensured that all the futures
+                        // completed successfully.
+                        throw new RuntimeException(e);
                     }
-                    return descriptions;
                 }
+                return descriptions;
             });
     }
 }

clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java

@@ -48,23 +48,13 @@ public KafkaFuture<Map<String, TopicListing>> namesToListings() {
      * Return a future which yields a collection of TopicListing objects.
      */
     public KafkaFuture<Collection<TopicListing>> listings() {
-        return future.thenApply(new KafkaFuture.BaseFunction<Map<String, TopicListing>, Collection<TopicListing>>() {
-            @Override
-            public Collection<TopicListing> apply(Map<String, TopicListing> namesToDescriptions) {
-                return namesToDescriptions.values();
-            }
-        });
+        return future.thenApply(namesToDescriptions -> namesToDescriptions.values());
     }
 
     /**
      * Return a future which yields a collection of topic names.
      */
     public KafkaFuture<Set<String>> names() {
-        return future.thenApply(new KafkaFuture.BaseFunction<Map<String, TopicListing>, Set<String>>() {
-            @Override
-            public Set<String> apply(Map<String, TopicListing> namesToListings) {
-                return namesToListings.keySet();
-            }
-        });
+        return future.thenApply(namesToListings -> namesToListings.keySet());
     }
 }

core/src/main/scala/kafka/controller/ControllerEventManager.scala

@@ -112,7 +112,7 @@ class ControllerEventManager(controllerId: Int,
 
   def isEmpty: Boolean = queue.isEmpty
 
-  class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptible = false) {
+  class ControllerEventThread(name: String) extends ShutdownableThread(name = name, isInterruptable = false) {
     logIdent = s"[ControllerEventThread controllerId=$controllerId] "
 
     override def doWork(): Unit = {

core/src/main/scala/kafka/log/LogCleaner.scala

@@ -269,7 +269,7 @@ class LogCleaner(initialConfig: CleanerConfig,
    * choosing the dirtiest log, cleaning it, and then swapping in the cleaned segments.
    */
   private[log] class CleanerThread(threadId: Int)
-    extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptible = false) {
+    extends ShutdownableThread(name = s"kafka-log-cleaner-thread-$threadId", isInterruptable = false) {
 
     protected override def loggerName = classOf[LogCleaner].getName
 

core/src/main/scala/kafka/utils/ShutdownableThread.scala

@@ -21,7 +21,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit}
 
 import org.apache.kafka.common.internals.FatalExitError
 
-abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean = true)
+abstract class ShutdownableThread(val name: String, val isInterruptable: Boolean = true)
         extends Thread(name) with Logging {
   this.setDaemon(false)
   this.logIdent = "[" + name + "]: "
@@ -50,7 +50,7 @@ abstract class ShutdownableThread(val name: String, val isInterruptible: Boolean
       if (isRunning) {
         info("Shutting down")
         shutdownInitiated.countDown()
-        if (isInterruptible)
+        if (isInterruptable)
           interrupt()
         true
       } else

core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala

@@ -1613,7 +1613,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
   }
 
   private class ProducerThread(clientId: String, retries: Int)
-    extends ShutdownableThread(clientId, isInterruptible = false) {
+    extends ShutdownableThread(clientId, isInterruptable = false) {
 
     private val producer = ProducerBuilder().maxRetries(retries).clientId(clientId).build()
     val lastSent = new ConcurrentHashMap[Int, Int]()
@@ -1634,7 +1634,7 @@ class DynamicBrokerReconfigurationTest extends ZooKeeperTestHarness with SaslSet
     }
   }
 
-  private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptible = false) {
+  private class ConsumerThread(producerThread: ProducerThread) extends ShutdownableThread("test-consumer", isInterruptable = false) {
     private val consumer = ConsumerBuilder("group1").enableAutoCommit(true).build()
     val lastReceived = new ConcurrentHashMap[Int, Int]()
     val missingRecords = new ConcurrentLinkedQueue[Int]()

core/src/test/scala/other/kafka/TestPurgatoryPerformance.scala

@@ -256,7 +256,7 @@ object TestPurgatoryPerformance {
 
   private class CompletionQueue {
     private[this] val delayQueue = new DelayQueue[Scheduled]()
-    private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptible = false) {
+    private[this] val thread = new ShutdownableThread(name = "completion thread", isInterruptable = false) {
       override def doWork(): Unit = {
         val scheduled = delayQueue.poll(100, TimeUnit.MILLISECONDS)
         if (scheduled != null) {

examples/README

@@ -6,4 +6,10 @@ To run the demo:
    2. For simple consumer demo, `run bin/java-simple-consumer-demo.sh`
    3. For unlimited sync-producer-consumer run, `run bin/java-producer-consumer-demo.sh sync`
    4. For unlimited async-producer-consumer run, `run bin/java-producer-consumer-demo.sh`
-
+   5. For standalone mode exactly once demo run, `run bin/exactly-once-demo.sh standaloneMode 6 3 50000`,
+      this means we are starting 3 EOS instances with 6 topic partitions and 50000 pre-populated records
+   6. For group mode exactly once demo run, `run bin/exactly-once-demo.sh groupMode 6 3 50000`,
+      this means the same as the standalone demo, except consumers are using subscription mode
+   7. Some notes for exactly once demo:
+      7.1. The Kafka server has to be on broker version 2.5 or higher to be able to run group mode.
+      7.2. You could also use Intellij to run the example directly by configuring parameters as "Program arguments"

examples/bin/exactly-once-demo.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (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.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+base_dir=$(dirname $0)/../..
+
+if [ "x$KAFKA_HEAP_OPTS" = "x" ]; then
+    export KAFKA_HEAP_OPTS="-Xmx512M"
+fi
+
+exec $base_dir/bin/kafka-run-class.sh kafka.examples.KafkaExactlyOnceDemo $@

examples/src/main/java/kafka/examples/Consumer.java

@@ -25,32 +25,56 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Properties;
+import java.util.concurrent.CountDownLatch;
 
 public class Consumer extends ShutdownableThread {
     private final KafkaConsumer<Integer, String> consumer;
     private final String topic;
+    private final int numMessageToConsume;
+    private int messageRemaining;
+    private final CountDownLatch latch;
 
-    public Consumer(String topic) {
+    public Consumer(final String topic,
+                    final String groupId,
+                    final boolean readCommitted,
+                    final int numMessageToConsume,
+                    final CountDownLatch latch) {
         super("KafkaConsumerExample", false);
         Properties props = new Properties();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT);
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "DemoConsumer");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
         props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
         props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerDeserializer");
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+        if (readCommitted) {
+            props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
+        }
+        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
 
         consumer = new KafkaConsumer<>(props);
         this.topic = topic;
+        this.numMessageToConsume = numMessageToConsume;
+        this.messageRemaining = numMessageToConsume;
+        this.latch = latch;
+    }
+
+    KafkaConsumer<Integer, String> get() {
+        return consumer;
     }
 
     @Override
     public void doWork() {
         consumer.subscribe(Collections.singletonList(this.topic));
         ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
         for (ConsumerRecord<Integer, String> record : records) {
-            System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
+            System.out.println("Verify-consumer received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
+        }
+        messageRemaining -= records.count();
+        if (messageRemaining <= 0) {
+            System.out.println("Verify-consumer finished reading " + numMessageToConsume + " messages");
+            latch.countDown();
         }
     }
 
@@ -60,7 +84,7 @@ public String name() {
     }
 
     @Override
-    public boolean isInterruptible() {
+    public boolean isInterruptable() {
         return false;
     }
 }