Skip to content

Commit c30d361

Browse files
davidmartinezbaruadavidmartinezbarua
authored
Merge pull request #15748 from amit88265/commit-offsets-in-kafka
BAEL-6946 added code for commit offsets
2 parents 8a2f421 + b57553e commit c30d361

File tree

9 files changed

+207
-0
lines changed

9 files changed

+207
-0
lines changed

‎apache-kafka-3/README.md

+9
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
## Apache Kafka
2+
3+
This module contains articles about Apache Kafka.
4+
5+
##### Building the project
6+
You can build the project from the command line using: *mvn clean install*, or in an IDE.
7+
8+
### Relevant Articles:
9+

‎apache-kafka-3/pom.xml

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<project xmlns="http://maven.apache.org/POM/4.0.0"
3+
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
4+
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
5+
<modelVersion>4.0.0</modelVersion>
6+
<artifactId>apache-kafka-3</artifactId>
7+
<name>apache-kafka-3</name>
8+
9+
<parent>
10+
<groupId>com.baeldung</groupId>
11+
<artifactId>parent-modules</artifactId>
12+
<version>1.0.0-SNAPSHOT</version>
13+
</parent>
14+
15+
<dependencies>
16+
<dependency>
17+
<groupId>org.apache.kafka</groupId>
18+
<artifactId>kafka-clients</artifactId>
19+
<version>${kafka.version}</version>
20+
</dependency>
21+
<dependency>
22+
<groupId>org.slf4j</groupId>
23+
<artifactId>slf4j-api</artifactId>
24+
<version>${org.slf4j.version}</version>
25+
</dependency>
26+
<dependency>
27+
<groupId>org.projectlombok</groupId>
28+
<artifactId>lombok</artifactId>
29+
<version>${lombok.version}</version>
30+
<scope>provided</scope>
31+
</dependency>
32+
<dependency>
33+
<groupId>com.fasterxml.jackson.core</groupId>
34+
<artifactId>jackson-databind</artifactId>
35+
<version>${jackson.databind.version}</version>
36+
</dependency>
37+
</dependencies>
38+
39+
<properties>
40+
<kafka.version>3.6.1</kafka.version>
41+
<jackson.databind.version>2.15.2</jackson.databind.version>
42+
</properties>
43+
<profiles>
44+
<profile>
45+
<id>integration-jdk9-and-above</id>
46+
</profile>
47+
</profiles>
48+
</project>

‎apache-kafka-3/src/main/java/com/baeldung/kafka/commitoffset/AsyncCommit.java

+23
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package com.baeldung.kafka.commitoffset;
2+
3+
import com.baeldung.kafka.commitoffset.config.KafkaConfigProperties;
4+
5+
import java.time.Duration;
6+
7+
import org.apache.kafka.clients.consumer.ConsumerRecord;
8+
import org.apache.kafka.clients.consumer.ConsumerRecords;
9+
import org.apache.kafka.clients.consumer.KafkaConsumer;
10+
11+
public class AsyncCommit {
12+
13+
public static void main(String[] args) {
14+
15+
KafkaConsumer<Long, String> consumer = new KafkaConsumer<>(KafkaConfigProperties.getProperties());
16+
consumer.subscribe(KafkaConfigProperties.getTopic());
17+
ConsumerRecords<Long, String> messages = consumer.poll(Duration.ofSeconds(10));
18+
for (ConsumerRecord<Long, String> message : messages) {
19+
// processed message
20+
consumer.commitAsync();
21+
}
22+
}
23+
}

‎apache-kafka-3/src/main/java/com/baeldung/kafka/commitoffset/AutomaticCommit.java

+26
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,26 @@
1+
package com.baeldung.kafka.commitoffset;
2+
3+
import com.baeldung.kafka.commitoffset.config.KafkaConfigProperties;
4+
5+
import java.time.Duration;
6+
import java.util.Properties;
7+
8+
import org.apache.kafka.clients.consumer.ConsumerConfig;
9+
import org.apache.kafka.clients.consumer.ConsumerRecord;
10+
import org.apache.kafka.clients.consumer.ConsumerRecords;
11+
import org.apache.kafka.clients.consumer.KafkaConsumer;
12+
13+
public class AutomaticCommit {
14+
15+
public static void main(String[] args) {
16+
17+
Properties properties = KafkaConfigProperties.getProperties();
18+
properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
19+
KafkaConsumer<Long, String> consumer = new KafkaConsumer<>(properties);
20+
consumer.subscribe(KafkaConfigProperties.getTopic());
21+
ConsumerRecords<Long, String> messages = consumer.poll(Duration.ofSeconds(10));
22+
for (ConsumerRecord<Long, String> message : messages) {
23+
// processed message
24+
}
25+
}
26+
}

‎apache-kafka-3/src/main/java/com/baeldung/kafka/commitoffset/SpecificOffsetCommit.java

+34
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,34 @@
1+
package com.baeldung.kafka.commitoffset;
2+
3+
import com.baeldung.kafka.commitoffset.config.KafkaConfigProperties;
4+
5+
import java.time.Duration;
6+
import java.util.HashMap;
7+
import java.util.Map;
8+
9+
import org.apache.kafka.clients.consumer.ConsumerRecord;
10+
import org.apache.kafka.clients.consumer.ConsumerRecords;
11+
import org.apache.kafka.clients.consumer.KafkaConsumer;
12+
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
13+
import org.apache.kafka.common.TopicPartition;
14+
15+
public class SpecificOffsetCommit {
16+
public static void main(String[] args) {
17+
18+
KafkaConsumer<Long, String> consumer = new KafkaConsumer<>(KafkaConfigProperties.getProperties());
19+
consumer.subscribe(KafkaConfigProperties.getTopic());
20+
Map<TopicPartition, OffsetAndMetadata> currentOffsets = new HashMap<>();
21+
int messageProcessed = 0;
22+
while (true) {
23+
ConsumerRecords<Long, String> messages = consumer.poll(Duration.ofSeconds(10));
24+
for (ConsumerRecord<Long, String> message : messages) {
25+
// processed message
26+
messageProcessed++;
27+
currentOffsets.put(new TopicPartition(message.topic(), message.partition()), new OffsetAndMetadata(message.offset() + 1));
28+
if (messageProcessed % 50 == 0) {
29+
consumer.commitSync(currentOffsets);
30+
}
31+
}
32+
}
33+
}
34+
}

‎apache-kafka-3/src/main/java/com/baeldung/kafka/commitoffset/SyncCommit.java

+23
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,23 @@
1+
package com.baeldung.kafka.commitoffset;
2+
3+
import com.baeldung.kafka.commitoffset.config.KafkaConfigProperties;
4+
5+
import java.time.Duration;
6+
7+
import org.apache.kafka.clients.consumer.ConsumerRecord;
8+
import org.apache.kafka.clients.consumer.ConsumerRecords;
9+
import org.apache.kafka.clients.consumer.KafkaConsumer;
10+
11+
public class SyncCommit {
12+
13+
public static void main(String[] args) {
14+
15+
KafkaConsumer<Long, String> consumer = new KafkaConsumer<>(KafkaConfigProperties.getProperties());
16+
consumer.subscribe(KafkaConfigProperties.getTopic());
17+
ConsumerRecords<Long, String> messages = consumer.poll(Duration.ofSeconds(10));
18+
for (ConsumerRecord<Long, String> message : messages) {
19+
// processed message
20+
consumer.commitSync();
21+
}
22+
}
23+
}

‎apache-kafka-3/src/main/java/com/baeldung/kafka/commitoffset/config/KafkaConfigProperties.java

+31
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,31 @@
1+
package com.baeldung.kafka.commitoffset.config;
2+
3+
import java.util.ArrayList;
4+
import java.util.Properties;
5+
6+
import org.apache.kafka.clients.consumer.ConsumerConfig;
7+
import org.apache.kafka.common.serialization.StringDeserializer;
8+
9+
/**
10+
* @author amitkumar
11+
*/
12+
public class KafkaConfigProperties {
13+
public static final String MY_TOPIC = "my-topic";
14+
15+
public static Properties getProperties() {
16+
17+
Properties props = new Properties();
18+
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
19+
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
20+
props.put(ConsumerConfig.GROUP_ID_CONFIG, "MyFirstConsumer");
21+
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
22+
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
23+
return props;
24+
}
25+
26+
public static ArrayList<String> getTopic() {
27+
ArrayList<String> topics = new ArrayList<>();
28+
topics.add(MY_TOPIC);
29+
return topics;
30+
}
31+
}

‎apache-kafka-3/src/test/resources/logback.xml

+11
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
<configuration>
2+
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
3+
<encoder>
4+
<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
5+
</encoder>
6+
</appender>
7+
8+
<root level="INFO">
9+
<appender-ref ref="STDOUT" />
10+
</root>
11+
</configuration>

‎pom.xml

+2
Original file line numberDiff line numberDiff line change
@@ -660,6 +660,7 @@
660660
<module>apache-httpclient4</module>
661661
<module>apache-httpclient</module>
662662
<module>apache-kafka-2</module>
663+
<module>apache-kafka-3</module>
663664
<module>apache-kafka</module>
664665
<module>apache-libraries-2</module>
665666
<module>apache-libraries</module>
@@ -907,6 +908,7 @@
907908
<module>apache-httpclient4</module>
908909
<module>apache-httpclient</module>
909910
<module>apache-kafka-2</module>
911+
<module>apache-kafka-3</module>
910912
<module>apache-kafka</module>
911913
<module>apache-libraries-2</module>
912914
<module>apache-libraries</module>

0 commit comments

Comments
 (0)