iSharkFly-Docs/java-tutorials
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

BAEL-6927: Update article "Understanding Kafka Topics and Partitions" (#14837)

Browse Source 
* consumer config. topic config. driver and calculator classes.

* basic app working. test structure

* final version

* wraping up

* optimizing imports

* comments changes

* addressing improvement changes
This commit is contained in:
Pedro Lopes 2023-10-01 12:22:34 -03:00 committed by GitHub
parent 0df6524e4a
commit ae29eca0f8
2 changed files with 6 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/TemperatureConsumer.java
View File

@ -8,29 +8,22 @@ import java.util.HashSet;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@Service @Service
public class TemperatureConsumer { public class TemperatureConsumer {
private CountDownLatch latch = new CountDownLatch(1);
Map<String, Set<String>> consumedRecords = new ConcurrentHashMap<>(); Map<String, Set<String>> consumedRecords = new ConcurrentHashMap<>();
@KafkaListener(topics = "celcius-scale-topic", groupId = "group-1") @KafkaListener(topics = "celcius-scale-topic", groupId = "group-1")
public void consumer1(ConsumerRecord<?, ?> consumerRecord) { public void consumer1(ConsumerRecord<?, ?> consumerRecord) {
computeConsumedRecord("consumer-1", consumerRecord.partition()); trackConsumedPartitions("consumer-1", consumerRecord.partition());
} }
private void computeConsumedRecord(String key, int consumerRecord) { private void trackConsumedPartitions(String consumerName, int partitionNumber) {
consumedRecords.computeIfAbsent(key, k -> new HashSet<>()); consumedRecords.computeIfAbsent(consumerName, k -> new HashSet<>());
consumedRecords.computeIfPresent(key, (k, v) -> { consumedRecords.computeIfPresent(consumerName, (k, v) -> {
v.add(String.valueOf(consumerRecord)); v.add(String.valueOf(partitionNumber));
return v; return v;
}); });
} }
public CountDownLatch getLatch() {
return latch;
}
} }

4
spring-kafka-2/src/test/java/com/baeldung/spring/kafka/topicsandpartitions/KafkaTopicsAndPartitionsIntegrationTest.java
View File

@ -7,8 +7,6 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka; import org.springframework.kafka.test.context.EmbeddedKafka;
import java.util.concurrent.TimeUnit;
@SpringBootTest(classes = ThermostatApplicationKafkaApp.class) @SpringBootTest(classes = ThermostatApplicationKafkaApp.class)
@EmbeddedKafka(partitions = 2, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"}) @EmbeddedKafka(partitions = 2, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
public class KafkaTopicsAndPartitionsIntegrationTest { public class KafkaTopicsAndPartitionsIntegrationTest {
@ -24,7 +22,7 @@ public class KafkaTopicsAndPartitionsIntegrationTest {
@Test @Test
public void givenTopic_andConsumerGroup_whenConsumersListenToEvents_thenConsumeItCorrectly() throws Exception { public void givenTopic_andConsumerGroup_whenConsumersListenToEvents_thenConsumeItCorrectly() throws Exception {
service.measureCelsiusAndPublish(10000); service.measureCelsiusAndPublish(10000);
consumer.getLatch().await(1, TimeUnit.SECONDS); Thread.sleep(1000);
System.out.println(consumer.consumedRecords); System.out.println(consumer.consumedRecords);
} }
} }