Following examples shows how to commit offset asynchronously with a callback and with the specified offset.
KafkaConsumer defines following method:
public void commitAsync(final Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback)
Where parameter 'offsets' is a map of offsets by partition with associate metadata. The callback is to recieve async commit callback (see last tutorial);
Example
Run Kafka server as described here
Example Config
package com.logicbig.example;
import java.util.Properties;
public class ExampleConfig {
public static final String BROKERS = "localhost:9092";
public static Properties getProducerProps() {
Properties props = new Properties();
props.put("bootstrap.servers", BROKERS);
props.put("acks", "all");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
return props;
}
public static Properties getConsumerProps() {
Properties props = new Properties();
props.setProperty("bootstrap.servers", BROKERS);
props.setProperty("group.id", "testGroup");
props.setProperty("enable.auto.commit", "false");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
return props;
}
}
Creating Example Topic
Let's create a topic with one partition
package com.logicbig.example;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import java.util.Collections;
import java.util.Properties;
import java.util.stream.Collectors;
public class TopicCreator {
public static void main(String[] args) throws Exception {
createTopic("example-topic-2020-5-28c", 1);
}
private static void createTopic(String topicName, int numPartitions) throws Exception {
Properties config = new Properties();
config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, ExampleConfig.BROKERS);
AdminClient admin = AdminClient.create(config);
//checking if topic already exists
boolean alreadyExists = admin.listTopics().names().get().stream()
.anyMatch(existingTopicName -> existingTopicName.equals(topicName));
if (alreadyExists) {
System.out.printf("topic already exits: %s%n", topicName);
} else {
//creating new topic
System.out.printf("creating topic: %s%n", topicName);
NewTopic newTopic = new NewTopic(topicName, numPartitions, (short) 1);
admin.createTopics(Collections.singleton(newTopic)).all().get();
}
//describing
System.out.println("-- describing topic --");
admin.describeTopics(Collections.singleton(topicName)).all().get()
.forEach((topic, desc) -> {
System.out.println("Topic: " + topic);
System.out.printf("Partitions: %s, partition ids: %s%n", desc.partitions().size(),
desc.partitions()
.stream()
.map(p -> Integer.toString(p.partition()))
.collect(Collectors.joining(",")));
});
admin.close();
}
}
creating topic: example-topic-2020-5-28c
-- describing topic --
Topic: example-topic-2020-5-28c
Partitions: 1, partition ids: 0
Using commitAsync(offsets, callback)
package com.logicbig.example;
import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import java.time.Duration;
import java.util.*;
public class CommitAsyncExample {
private static String TOPIC_NAME = "example-topic-2020-5-28c";
private static KafkaConsumer<String, String> consumer;
private static TopicPartition topicPartition;
public static void main(String[] args) throws Exception {
Properties consumerProps = ExampleConfig.getConsumerProps();
consumer = new KafkaConsumer<>(consumerProps);
topicPartition = new TopicPartition(TOPIC_NAME, 0);
consumer.assign(Collections.singleton(topicPartition));
printOffsets("before consumer loop");
sendMessages();
startConsumer();
}
private static void startConsumer() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
for (ConsumerRecord<String, String> record : records) {
System.out.printf("consumed: key = %s, value = %s, partition id= %s, offset = %s%n",
record.key(), record.value(), record.partition(), record.offset());
consumer.commitAsync(
Collections.singletonMap(topicPartition, new OffsetAndMetadata(record.offset())),
(offsets, exception) -> {
System.out.printf("Callback, offset: %s, exception %s%n", offsets, exception);
});
}
if (records.isEmpty()) {
System.out.println("-- terminating consumer --");
break;
}
}
printOffsets("after consumer loop");
}
private static void printOffsets(String message) {
Map<TopicPartition, OffsetAndMetadata> committed = consumer
.committed(new HashSet<>(Arrays.asList(topicPartition)));
OffsetAndMetadata offsetAndMetadata = committed.get(topicPartition);
long position = consumer.position(topicPartition);
System.out
.printf("Offset info %s, Committed: %s, current position %s%n", message,
offsetAndMetadata == null ? null : offsetAndMetadata
.offset(), position);
}
private static void sendMessages() {
Properties producerProps = ExampleConfig.getProducerProps();
KafkaProducer producer = new KafkaProducer<>(producerProps);
for (int i = 0; i < 4; i++) {
String value = "message-" + i;
System.out.printf("Sending message topic: %s, value: %s%n", TOPIC_NAME, value);
producer.send(new ProducerRecord<>(TOPIC_NAME, value));
}
producer.flush();
producer.close();
}
}
Offset info before consumer loop, Committed: null, current position 0
Sending message topic: example-topic-2020-5-28c, value: message-0
Sending message topic: example-topic-2020-5-28c, value: message-1
Sending message topic: example-topic-2020-5-28c, value: message-2
Sending message topic: example-topic-2020-5-28c, value: message-3
consumed: key = null, value = message-0, partition id= 0, offset = 0
consumed: key = null, value = message-1, partition id= 0, offset = 1
consumed: key = null, value = message-2, partition id= 0, offset = 2
consumed: key = null, value = message-3, partition id= 0, offset = 3
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=0, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=1, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=2, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=3, leaderEpoch=null, metadata=''}}, exception null
-- terminating consumer --
Offset info after consumer loop, Committed: 3, current position 4
Running one more time
Offset info before consumer loop, Committed: 3, current position 3
Sending message topic: example-topic-2020-5-28c, value: message-0
Sending message topic: example-topic-2020-5-28c, value: message-1
Sending message topic: example-topic-2020-5-28c, value: message-2
Sending message topic: example-topic-2020-5-28c, value: message-3
consumed: key = null, value = message-3, partition id= 0, offset = 3
consumed: key = null, value = message-0, partition id= 0, offset = 4
consumed: key = null, value = message-1, partition id= 0, offset = 5
consumed: key = null, value = message-2, partition id= 0, offset = 6
consumed: key = null, value = message-3, partition id= 0, offset = 7
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=3, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=4, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=5, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=6, leaderEpoch=null, metadata=''}}, exception null
Callback, offset: {example-topic-2020-5-28c-0=OffsetAndMetadata{offset=7, leaderEpoch=null, metadata=''}}, exception null
-- terminating consumer --
Offset info after consumer loop, Committed: 7, current position 8
Example ProjectDependencies and Technologies Used: - kafka_2.12 2.5.0 Apache Kafka
- JDK 8
- Maven 3.5.4
|
|
