diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java index 9822b423c4d82..ffb6470e503c5 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/DescribeTopicsResult.java @@ -51,21 +51,18 @@ public Map> values() { */ public KafkaFuture> all() { return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0])). - thenApply(new KafkaFuture.BaseFunction>() { - @Override - public Map apply(Void v) { - Map descriptions = new HashMap<>(futures.size()); - for (Map.Entry> 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 descriptions = new HashMap<>(futures.size()); + for (Map.Entry> 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; }); } } diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java index 4e7e1a2e98721..21540732d30d2 100644 --- a/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java +++ b/clients/src/main/java/org/apache/kafka/clients/admin/ListTopicsResult.java @@ -48,23 +48,13 @@ public KafkaFuture> namesToListings() { * Return a future which yields a collection of TopicListing objects. */ public KafkaFuture> listings() { - return future.thenApply(new KafkaFuture.BaseFunction, Collection>() { - @Override - public Collection apply(Map namesToDescriptions) { - return namesToDescriptions.values(); - } - }); + return future.thenApply(namesToDescriptions -> namesToDescriptions.values()); } /** * Return a future which yields a collection of topic names. */ public KafkaFuture> names() { - return future.thenApply(new KafkaFuture.BaseFunction, Set>() { - @Override - public Set apply(Map namesToListings) { - return namesToListings.keySet(); - } - }); + return future.thenApply(namesToListings -> namesToListings.keySet()); } } diff --git a/examples/README b/examples/README index f6e3410a2a0aa..2efe71ac182ee 100644 --- a/examples/README +++ b/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" diff --git a/examples/bin/exactly-once-demo.sh b/examples/bin/exactly-once-demo.sh new file mode 100755 index 0000000000000..e9faa428d3f27 --- /dev/null +++ b/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 $@ diff --git a/examples/src/main/java/kafka/examples/Consumer.java b/examples/src/main/java/kafka/examples/Consumer.java index 26d6e23a3f8e8..19cb67ccc9158 100644 --- a/examples/src/main/java/kafka/examples/Consumer.java +++ b/examples/src/main/java/kafka/examples/Consumer.java @@ -25,24 +25,45 @@ 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 consumer; private final String topic; + private final String groupId; + 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); + this.groupId = groupId; 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 get() { + return consumer; } @Override @@ -50,7 +71,12 @@ public void doWork() { consumer.subscribe(Collections.singletonList(this.topic)); ConsumerRecords records = consumer.poll(Duration.ofSeconds(1)); for (ConsumerRecord record : records) { - System.out.println("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); + System.out.println(groupId + " received message : from partition " + record.partition() + ", (" + record.key() + ", " + record.value() + ") at offset " + record.offset()); + } + messageRemaining -= records.count(); + if (messageRemaining <= 0) { + System.out.println(groupId + " finished reading " + numMessageToConsume + " messages"); + latch.countDown(); } } diff --git a/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java new file mode 100644 index 0000000000000..53685f351503d --- /dev/null +++ b/examples/src/main/java/kafka/examples/ExactlyOnceMessageProcessor.java @@ -0,0 +1,209 @@ +/* + * 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. + */ +package kafka.examples; + +import org.apache.kafka.clients.consumer.CommitFailedException; +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetAndMetadata; +import org.apache.kafka.clients.producer.KafkaProducer; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.FencedInstanceIdException; +import org.apache.kafka.common.errors.ProducerFencedException; + +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicLong; + +/** + * A demo class for how to write a customized EOS app. It takes a consume-process-produce loop. + * Important configurations and APIs are commented. + */ +public class ExactlyOnceMessageProcessor extends Thread { + + private static final boolean READ_COMMITTED = true; + + private final String mode; + private final String inputTopic; + private final String outputTopic; + private final String consumerGroupId; + private final int numPartitions; + private final int numInstances; + private final int instanceIdx; + private final String transactionalId; + + private final KafkaProducer producer; + private final KafkaConsumer consumer; + + private final CountDownLatch latch; + + public ExactlyOnceMessageProcessor(final String mode, + final String inputTopic, + final String outputTopic, + final int numPartitions, + final int numInstances, + final int instanceIdx, + final CountDownLatch latch) { + this.mode = mode; + this.inputTopic = inputTopic; + this.outputTopic = outputTopic; + this.consumerGroupId = "Eos-consumer"; + this.numPartitions = numPartitions; + this.numInstances = numInstances; + this.instanceIdx = instanceIdx; + this.transactionalId = "Processor-" + instanceIdx; + // A unique transactional.id must be provided in order to properly use EOS. + producer = new Producer(outputTopic, true, transactionalId, true, -1, null).get(); + // Consumer must be in read_committed mode, which means it won't be able to read uncommitted data. + consumer = new Consumer(inputTopic, consumerGroupId, READ_COMMITTED, -1, null).get(); + this.latch = latch; + } + + @Override + public void run() { + // Init transactions call should always happen first in order to clear zombie transactions from previous generation. + producer.initTransactions(); + + final AtomicLong messageRemaining = new AtomicLong(Long.MAX_VALUE); + + // Under group mode, topic based subscription is sufficient as EOS apps are safe to cooperate transactionally after 2.5. + // Under standalone mode, user needs to manually assign the topic partitions and make sure the assignment is unique + // across the consumer group instances. + if (this.mode.equals("groupMode")) { + consumer.subscribe(Collections.singleton(inputTopic), new ConsumerRebalanceListener() { + @Override + public void onPartitionsRevoked(Collection partitions) { + printWithTxnId("Revoked partition assignment to kick-off rebalancing: " + partitions); + } + + @Override + public void onPartitionsAssigned(Collection partitions) { + printWithTxnId("Received partition assignment after rebalancing: " + partitions); + messageRemaining.set(messagesRemaining(consumer)); + } + }); + } else { + // Do a range assignment of topic partitions. + List topicPartitions = new ArrayList<>(); + int rangeSize = numPartitions / numInstances; + int startPartition = rangeSize * instanceIdx; + int endPartition = Math.min(numPartitions - 1, startPartition + rangeSize - 1); + for (int partition = startPartition; partition <= endPartition; partition++) { + topicPartitions.add(new TopicPartition(inputTopic, partition)); + } + + consumer.assign(topicPartitions); + printWithTxnId("Manually assign partitions: " + topicPartitions); + } + + int messageProcessed = 0; + boolean abortPreviousTransaction = false; + while (messageRemaining.get() > 0) { + ConsumerRecords records = consumer.poll(Duration.ofMillis(200)); + if (records.count() > 0) { + try { + // Abort previous transaction if instructed. + if (abortPreviousTransaction) { + producer.abortTransaction(); + // The consumer fetch position also needs to be reset. + resetToLastCommittedPositions(consumer); + abortPreviousTransaction = false; + } + // Begin a new transaction session. + producer.beginTransaction(); + for (ConsumerRecord record : records) { + // Process the record and send to downstream. + ProducerRecord customizedRecord = transform(record); + producer.send(customizedRecord); + } + Map positions = new HashMap<>(); + for (TopicPartition topicPartition : consumer.assignment()) { + positions.put(topicPartition, new OffsetAndMetadata(consumer.position(topicPartition), null)); + } + // Checkpoint the progress by sending offsets to group coordinator broker. + // Under group mode, we must apply consumer group metadata for proper fencing. + if (this.mode.equals("groupMode")) { + producer.sendOffsetsToTransaction(positions, consumer.groupMetadata()); + } else { + producer.sendOffsetsToTransaction(positions, consumerGroupId); + } + + // Finish the transaction. All sent records should be visible for consumption now. + producer.commitTransaction(); + messageProcessed += records.count(); + } catch (CommitFailedException e) { + // In case of a retriable exception, suggest aborting the ongoing transaction for correctness. + abortPreviousTransaction = true; + } catch (ProducerFencedException | FencedInstanceIdException e) { + throw new KafkaException("Encountered fatal error during processing: " + e.getMessage()); + } + } + messageRemaining.set(messagesRemaining(consumer)); + printWithTxnId("Message remaining: " + messageRemaining); + } + + printWithTxnId("Finished processing " + messageProcessed + " records"); + latch.countDown(); + } + + private void printWithTxnId(final String message) { + System.out.println(transactionalId + ": " + message); + } + + private ProducerRecord transform(final ConsumerRecord record) { + printWithTxnId("Transformed record (" + record.key() + "," + record.value() + ")"); + return new ProducerRecord<>(outputTopic, record.key() / 2, "Transformed_" + record.value()); + } + + private long messagesRemaining(final KafkaConsumer consumer) { + final Map fullEndOffsets = consumer.endOffsets(new ArrayList<>(consumer.assignment())); + // If we couldn't detect any end offset, that means we are still not able to fetch offsets. + if (fullEndOffsets.isEmpty()) { + return Long.MAX_VALUE; + } + + return consumer.assignment().stream().mapToLong(partition -> { + long currentPosition = consumer.position(partition); + printWithTxnId("Processing partition " + partition + " with full offsets " + fullEndOffsets); + if (fullEndOffsets.containsKey(partition)) { + return fullEndOffsets.get(partition) - currentPosition; + } + return 0; + }).sum(); + } + + private static void resetToLastCommittedPositions(KafkaConsumer consumer) { + final Map committed = consumer.committed(consumer.assignment()); + consumer.assignment().forEach(tp -> { + OffsetAndMetadata offsetAndMetadata = committed.get(tp); + if (offsetAndMetadata != null) + consumer.seek(tp, offsetAndMetadata.offset()); + else + consumer.seekToBeginning(Collections.singleton(tp)); + }); + } +} diff --git a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java index f42ed6f4abd1e..561732bec6c1a 100644 --- a/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java +++ b/examples/src/main/java/kafka/examples/KafkaConsumerProducerDemo.java @@ -16,14 +16,20 @@ */ package kafka.examples; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + public class KafkaConsumerProducerDemo { - public static void main(String[] args) { + public static void main(String[] args) throws InterruptedException { boolean isAsync = args.length == 0 || !args[0].trim().equalsIgnoreCase("sync"); - Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync); + CountDownLatch latch = new CountDownLatch(2); + Producer producerThread = new Producer(KafkaProperties.TOPIC, isAsync, null, false, 10000, latch); producerThread.start(); - Consumer consumerThread = new Consumer(KafkaProperties.TOPIC); + Consumer consumerThread = new Consumer(KafkaProperties.TOPIC, "DemoConsumer", false, 10000, latch); consumerThread.start(); - + latch.await(5, TimeUnit.MINUTES); + consumerThread.shutdown(); + System.out.println("All finished!"); } } diff --git a/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java new file mode 100644 index 0000000000000..d418ebaa88541 --- /dev/null +++ b/examples/src/main/java/kafka/examples/KafkaExactlyOnceDemo.java @@ -0,0 +1,185 @@ +/* + * 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. + */ +package kafka.examples; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.errors.TopicExistsException; +import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; + +import java.util.Arrays; +import java.util.List; +import java.util.Properties; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +/** + * This exactly once demo driver takes 4 arguments: + * - mode: whether to run as standalone app, or a group + * - partition: number of partitions for input/output topic + * - instances: number of instances + * - records: number of records + * An example argument list would be `groupMode 6 3 50000` + * + * The driver could be decomposed as following stages: + * + * 1. Cleanup any topic whose name conflicts with input and output topic, so that we have a clean-start. + * + * 2. Set up a producer in a separate thread to pre-populate a set of records with even number keys into + * the input topic. The driver will block for the record generation to finish, so the producer + * must be in synchronous sending mode. + * + * 3. Set up transactional instances in separate threads which does a consume-process-produce loop, + * tailing data from input topic (See {@link ExactlyOnceMessageProcessor}). Each EOS instance will + * drain all the records from either given partitions or auto assigned partitions by actively + * comparing log end offset with committed offset. Each record will be processed exactly once + * as dividing the key by 2, and extend the value message. The driver will block for all the record + * processing to finish. The transformed record shall be written to the output topic, with + * transactional guarantee. + * + * 4. Set up a read committed consumer in a separate thread to verify we have all records within + * the output topic, while the message ordering on partition level is maintained. + * The driver will block for the consumption of all committed records. + * + * From this demo, you could see that all the records from pre-population are processed exactly once, + * in either standalone mode or group mode, with strong partition level ordering guarantee. + * + * Note: please start the kafka broker and zookeeper in local first. The broker version must be >= 2.5 + * in order to run group mode, otherwise the app could throw + * {@link org.apache.kafka.common.errors.UnsupportedVersionException}. + */ +public class KafkaExactlyOnceDemo { + + private static final String INPUT_TOPIC = "input-topic"; + private static final String OUTPUT_TOPIC = "output-topic"; + + public static void main(String[] args) throws InterruptedException, ExecutionException { + if (args.length != 4) { + throw new IllegalArgumentException("Should accept 4 parameters: [mode], " + + "[number of partitions], [number of instances], [number of records]"); + } + + String mode = args[0]; + int numPartitions = Integer.valueOf(args[1]); + int numInstances = Integer.valueOf(args[2]); + int numRecords = Integer.valueOf(args[3]); + + /* Stage 1: topic cleanup and recreation */ + recreateTopics(numPartitions); + + CountDownLatch prePopulateLatch = new CountDownLatch(1); + + /* Stage 2: pre-populate records */ + Producer producerThread = new Producer(INPUT_TOPIC, false, null, true, numRecords, prePopulateLatch); + producerThread.start(); + + prePopulateLatch.await(5, TimeUnit.MINUTES); + + CountDownLatch transactionalCopyLatch = new CountDownLatch(numInstances); + + /* Stage 3: transactionally process all messages */ + for (int instanceIdx = 0; instanceIdx < numInstances; instanceIdx++) { + ExactlyOnceMessageProcessor messageProcessor = new ExactlyOnceMessageProcessor(mode, + INPUT_TOPIC, OUTPUT_TOPIC, numPartitions, + numInstances, instanceIdx, transactionalCopyLatch); + messageProcessor.start(); + } + + transactionalCopyLatch.await(5, TimeUnit.MINUTES); + + CountDownLatch consumeLatch = new CountDownLatch(1); + + /* Stage 4: consume all processed messages to verify exactly once */ + Consumer consumerThread = new Consumer(OUTPUT_TOPIC, "Verify-consumer", true, numRecords, consumeLatch); + consumerThread.start(); + + consumeLatch.await(5, TimeUnit.MINUTES); + consumerThread.shutdown(); + System.out.println("All finished!"); + } + + private static void recreateTopics(final int numPartitions) + throws ExecutionException, InterruptedException { + Properties props = new Properties(); + props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, + KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); + + Admin adminClient = Admin.create(props); + + List topicsToDelete = Arrays.asList(INPUT_TOPIC, OUTPUT_TOPIC); + + deleteTopic(adminClient, topicsToDelete); + + // Check topic existence in a retry loop + while (true) { + System.out.println("Making sure the topics are deleted successfully: " + topicsToDelete); + + Set listedTopics = adminClient.listTopics().names().get(); + System.out.println("Current list of topics: " + listedTopics); + + boolean hasTopicInfo = false; + for (String listedTopic : listedTopics) { + if (topicsToDelete.contains(listedTopic)) { + hasTopicInfo = true; + break; + } + } + if (!hasTopicInfo) { + break; + } + Thread.sleep(1000); + } + + // Create topics in a retry loop + while (true) { + final short replicationFactor = 1; + final List newTopics = Arrays.asList( + new NewTopic(INPUT_TOPIC, numPartitions, replicationFactor), + new NewTopic(OUTPUT_TOPIC, numPartitions, replicationFactor)); + try { + adminClient.createTopics(newTopics).all().get(); + System.out.println("Created new topics: " + newTopics); + break; + } catch (ExecutionException e) { + if (!(e.getCause() instanceof TopicExistsException)) { + throw e; + } + System.out.println("Metadata of the old topics are not cleared yet..."); + + deleteTopic(adminClient, topicsToDelete); + + Thread.sleep(1000); + } + } + } + + private static void deleteTopic(final Admin adminClient, final List topicsToDelete) + throws InterruptedException, ExecutionException { + try { + adminClient.deleteTopics(topicsToDelete).all().get(); + } catch (ExecutionException e) { + if (!(e.getCause() instanceof UnknownTopicOrPartitionException)) { + throw e; + } + System.out.println("Encountered exception during topic deletion: " + e.getCause()); + } + System.out.println("Deleted old topics: " + topicsToDelete); + } +} diff --git a/examples/src/main/java/kafka/examples/Producer.java b/examples/src/main/java/kafka/examples/Producer.java index b6998c58ac756..3805dd389a4f5 100644 --- a/examples/src/main/java/kafka/examples/Producer.java +++ b/examples/src/main/java/kafka/examples/Producer.java @@ -25,45 +25,69 @@ import org.apache.kafka.common.serialization.StringSerializer; import java.util.Properties; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; public class Producer extends Thread { private final KafkaProducer producer; private final String topic; private final Boolean isAsync; + private int numRecords; + private final CountDownLatch latch; - public Producer(String topic, Boolean isAsync) { + public Producer(final String topic, + final Boolean isAsync, + final String transactionalId, + final boolean enableIdempotency, + final int numRecords, + final CountDownLatch latch) { Properties props = new Properties(); props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaProperties.KAFKA_SERVER_URL + ":" + KafkaProperties.KAFKA_SERVER_PORT); props.put(ProducerConfig.CLIENT_ID_CONFIG, "DemoProducer"); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class.getName()); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); + if (transactionalId != null) { + props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId); + } + props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, enableIdempotency); + producer = new KafkaProducer<>(props); this.topic = topic; this.isAsync = isAsync; + this.numRecords = numRecords; + this.latch = latch; + } + + KafkaProducer get() { + return producer; } + @Override public void run() { - int messageNo = 1; - while (true) { - String messageStr = "Message_" + messageNo; + int messageKey = 0; + int recordsSent = 0; + while (recordsSent < numRecords) { + String messageStr = "Message_" + messageKey; long startTime = System.currentTimeMillis(); if (isAsync) { // Send asynchronously producer.send(new ProducerRecord<>(topic, - messageNo, - messageStr), new DemoCallBack(startTime, messageNo, messageStr)); + messageKey, + messageStr), new DemoCallBack(startTime, messageKey, messageStr)); } else { // Send synchronously try { producer.send(new ProducerRecord<>(topic, - messageNo, + messageKey, messageStr)).get(); - System.out.println("Sent message: (" + messageNo + ", " + messageStr + ")"); + System.out.println("Sent message: (" + messageKey + ", " + messageStr + ")"); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } } - ++messageNo; + messageKey += 2; + recordsSent += 1; } + System.out.println("Producer sent " + numRecords + " records successfully"); + latch.countDown(); } }