-
-
Notifications
You must be signed in to change notification settings - Fork 216
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #1283 from fdelbrayelle/feat/issues/1096_kafka_con…
…sumer Feat/issues/1096 kafka consumer
- Loading branch information
Showing
20 changed files
with
512 additions
and
158 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
65 changes: 65 additions & 0 deletions
65
src/main/resources/generator/server/springboot/broker/kafka/AbstractConsumer.java.mustache
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| package {{packageName}}.dummy.infrastructure.primary.kafka.consumer; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.Collections; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
| 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.common.errors.WakeupException; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public abstract class AbstractConsumer<T> implements Runnable { | ||
| private final Logger log = LoggerFactory.getLogger(AbstractConsumer.class); | ||
| private final AtomicBoolean closed = new AtomicBoolean(false); | ||
| private final KafkaConsumer<String, T> consumer; | ||
| private final String topicName; | ||
| private final int pollingTimeout; | ||
| protected AbstractConsumer(final String topicName, final int pollingTimeout, final KafkaConsumer<String, T> consumer) { | ||
| this.topicName = topicName; | ||
| this.pollingTimeout = pollingTimeout; | ||
| this.consumer = consumer; | ||
| } | ||
|
|
||
| @Override | ||
| public void run() { | ||
| try { | ||
| consumer.subscribe(Collections.singleton(topicName)); | ||
| while (!closed.get()) { | ||
| final ConsumerRecords<String, T> records = consumer.poll(Duration.ofMillis(pollingTimeout)); | ||
| records.forEach(this::handleMessage); | ||
| consumer.commitSync(); | ||
| } | ||
| } catch (final WakeupException e) { | ||
| // Ignore exception if closing | ||
| if (!closed.get()) { | ||
| throw e; | ||
| } | ||
| } catch (final Exception e) { | ||
| log.error("An error occurred while trying to poll records from topic!", e); | ||
| } finally { | ||
| consumer.close(); | ||
| } | ||
| } | ||
|
|
||
| // Shutdown hook which can be called from a separate thread | ||
| public void shutdown() { | ||
| closed.set(true); | ||
| consumer.wakeup(); | ||
| } | ||
|
|
||
| public boolean isClosed() { | ||
| return closed.get(); | ||
| } | ||
|
|
||
| public void setClosed(boolean closed) { | ||
| this.closed.set(closed); | ||
| } | ||
|
|
||
| protected abstract boolean handleMessage(ConsumerRecord<String, T> consumerRecord); | ||
| } |
57 changes: 57 additions & 0 deletions
57
src/main/resources/generator/server/springboot/broker/kafka/DummyConsumer.java.mustache
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,57 @@ | ||
| package {{packageName}}.dummy.infrastructure.primary.kafka.consumer; | ||
|
|
||
| import javax.annotation.PostConstruct; | ||
| import javax.annotation.PreDestroy; | ||
| import org.apache.kafka.clients.consumer.ConsumerRecord; | ||
| import org.apache.kafka.clients.consumer.KafkaConsumer; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.springframework.beans.factory.annotation.Value; | ||
| import org.springframework.context.annotation.Bean; | ||
| import org.springframework.core.task.SimpleAsyncTaskExecutor; | ||
| import org.springframework.stereotype.Service; | ||
|
|
||
| @Service | ||
| public class DummyConsumer extends AbstractConsumer<String> { | ||
| private final Logger log = LoggerFactory.getLogger(DummyConsumer.class); | ||
| public DummyConsumer( | ||
| @Value("${kafka.topic.dummy}") final String topicName, | ||
| @Value("${kafka.polling.timeout}") final int pollingTimeout, | ||
| final KafkaConsumer<String, String> kafkaConsumer | ||
| ) { | ||
| super(topicName, pollingTimeout, kafkaConsumer); | ||
| } | ||
|
|
||
| @PostConstruct | ||
| public void init() { | ||
| Runtime.getRuntime().addShutdownHook(new Thread(this::shutdown)); | ||
| } | ||
|
|
||
| @PreDestroy | ||
| public void destroy() { | ||
| shutdown(); | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean handleMessage(final ConsumerRecord<String, String> consumerRecord) { | ||
| // /!\ Maybe you could delete the next log calls to avoid disclosing personal user information | ||
| final String value = consumerRecord.value(); | ||
| if (value == null) { | ||
| log.error("Null value in record {}", consumerRecord); | ||
| return false; | ||
| } | ||
|
|
||
| log.info("Handling record: {}", value); | ||
| // Here is where you can handle your records | ||
| return true; | ||
| } | ||
|
|
||
| @Bean | ||
| public void executeKafkaRunner() { | ||
| new SimpleAsyncTaskExecutor().execute(this); | ||
| } | ||
| } |
30 changes: 30 additions & 0 deletions
30
src/main/resources/generator/server/springboot/broker/kafka/DummyConsumerIT.java.mustache
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,30 @@ | ||
| package {{packageName}}.dummy.infrastructure.primary.kafka.consumer; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerRecord; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.springframework.beans.factory.annotation.Autowired; | ||
| import {{packageName}}.IntegrationTest; | ||
| import {{packageName}}.dummy.infrastructure.secondary.kafka.producer.DummyProducer; | ||
|
|
||
| @IntegrationTest | ||
| class DummyConsumerIT { | ||
| @Autowired | ||
| private DummyProducer dummyProducer; | ||
| @Autowired | ||
| private DummyConsumer dummyConsumer; | ||
| @Test | ||
| void shouldHandleMessage() { | ||
| final String messageToSend = "dummy"; | ||
| dummyProducer.send(messageToSend); | ||
| ConsumerRecord<String, String> record = new ConsumerRecord<>("queue.kafkaapp.dummy", 0, 0, null, messageToSend); | ||
| boolean actualResult = dummyConsumer.handleMessage(record); | ||
| assertThat(actualResult).isTrue(); | ||
| } | ||
| } |
Oops, something went wrong.