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

BAEL-6406 - Understanding Kafka Topics and Partitions (#14324) 

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

* basic app working. test structure

* final version

* wraping up

* optimizing imports
This commit is contained in:
Pedro Lopes 2023-08-01 00:01:26 -03:00 committed by GitHub
parent c8bca3abc1
commit 5843e8212a
7 changed files with 207 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/KafkaConsumerConfig.java Normal file
View File

@ -0,0 +1,39 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.DoubleDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import java.util.HashMap;
import java.util.Map;
@EnableKafka
@Configuration
public class KafkaConsumerConfig {
@Value(value = "${spring.kafka.bootstrap-servers}")
private String bootstrapAddress;
@Bean
public ConsumerFactory<String, Double> consumerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, DoubleDeserializer.class);
return new DefaultKafkaConsumerFactory<>(props);
}
@Bean
public ConcurrentKafkaListenerContainerFactory<String, Double> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, Double> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
return factory;
}
}

35
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/KafkaProducerConfig.java Normal file
View File

@ -0,0 +1,35 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.DoubleSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaProducerConfig {
@Value(value = "${spring.kafka.bootstrap-servers}")
private String bootstrapAddress;
@Bean
public ProducerFactory<String, Double> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, DoubleSerializer.class);
return new DefaultKafkaProducerFactory<>(configProps);
}
@Bean
public KafkaTemplate<String, Double> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}

30
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/KafkaTopicConfig.java Normal file
View File

@ -0,0 +1,30 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewTopic;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.core.KafkaAdmin;
import java.util.HashMap;
import java.util.Map;
@Configuration
public class KafkaTopicConfig {
@Value(value = "${spring.kafka.bootstrap-servers}")
private String bootstrapAddress;
public KafkaAdmin kafkaAdmin() {
Map<String, Object> configs = new HashMap<>();
configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
return new KafkaAdmin(configs);
}
public NewTopic celciusTopic() {
return TopicBuilder.name("celcius-scale-topic")
.partitions(2)
.build();
}
}

36
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/TemperatureConsumer.java Normal file
View File

@ -0,0 +1,36 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
@Service
public class TemperatureConsumer {
private CountDownLatch latch = new CountDownLatch(1);
Map<String, Set<String>> consumedRecords = new ConcurrentHashMap<>();
@KafkaListener(topics = "celcius-scale-topic", groupId = "group-1")
public void consumer1(ConsumerRecord<?, ?> consumerRecord) {
computeConsumedRecord("consumer-1", consumerRecord.partition());
}
private void computeConsumedRecord(String key, int consumerRecord) {
consumedRecords.computeIfAbsent(key, k -> new HashSet<>());
consumedRecords.computeIfPresent(key, (k, v) -> {
v.add(String.valueOf(consumerRecord));
return v;
});
}
public CountDownLatch getLatch() {
return latch;
}
}

13
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/ThermostatApplicationKafkaApp.java Normal file
View File

@ -0,0 +1,13 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Import;
@SpringBootApplication
@Import(value = {KafkaTopicConfig.class, KafkaProducerConfig.class, KafkaConsumerConfig.class})
public class ThermostatApplicationKafkaApp {
public static void main(String[] args) {
SpringApplication.run(ThermostatApplicationKafkaApp.class, args);
}
}

24
spring-kafka-2/src/main/java/com/baeldung/spring/kafka/topicsandpartitions/ThermostatService.java Normal file
View File

@ -0,0 +1,24 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import java.util.Random;
@Service
public class ThermostatService {
private final KafkaTemplate<String, Double> kafkaTemplate;
public ThermostatService(KafkaTemplate<String, Double> kafkaTemplate) {
this.kafkaTemplate = kafkaTemplate;
}
public void measureCelsiusAndPublish(int numMeasurements) {
new Random().doubles(25, 35)
.limit(numMeasurements)
.forEach(tmp -> {
kafkaTemplate.send("celcius-scale-topic", tmp);
});
}
}

30
spring-kafka-2/src/test/java/com/baeldung/spring/kafka/topicsandpartitions/KafkaTopicsAndPartitionsIntegrationTest.java Normal file
View File

@ -0,0 +1,30 @@
package com.baeldung.spring.kafka.topicsandpartitions;
import org.junit.ClassRule;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.test.EmbeddedKafkaBroker;
import org.springframework.kafka.test.context.EmbeddedKafka;
import java.util.concurrent.TimeUnit;
@SpringBootTest(classes = ThermostatApplicationKafkaApp.class)
@EmbeddedKafka(partitions = 2, brokerProperties = {"listeners=PLAINTEXT://localhost:9092", "port=9092"})
public class KafkaTopicsAndPartitionsIntegrationTest {
@ClassRule
public static EmbeddedKafkaBroker embeddedKafka = new EmbeddedKafkaBroker(1, true, "multitype");
@Autowired
private ThermostatService service;
@Autowired
private TemperatureConsumer consumer;
@Test
public void givenTopic_andConsumerGroup_whenConsumersListenToEvents_thenConsumeItCorrectly() throws Exception {
service.measureCelsiusAndPublish(10000);
consumer.getLatch().await(1, TimeUnit.SECONDS);
System.out.println(consumer.consumedRecords);
}
}