Extending KafkaConsumer

Extending springframework.kafka We’re already using org.springframework.kafka but not taking advantage of its concurrent features and not taking advantage of features that allow for pausing and starting consumers.

Our implementation of InitialContributionJobMessageConsumer could contain a single reference to the ConcurrentMessageListenerContainer. Our message processing code could easily be injected into the implementation of an AcknowledgingMessageListener.

When a message cannot be committed because the central server is unavailable, we could pause the ConcurrentMessageListenerContainer, and schedule a retry using a ConcurrentTaskScheduler. · Javadoc for ConcurrentMessageListenerContainer

·Example code using ConcurrentMessageListenerContainer to read messages form kafka with pause and restart: ·

Medium post describing the above repo

Pausing and resuming using CurrenMessageListenerContainer

Using Spring Batch. Spring Batch does much more than read messages from kafka, but it can easily be adapted to perform that task. See the notes for an example for how to read from kafka. The reason we should strongly consider Spring Batch is that unlike extending springframework.kafka we get job persistence for free in the form of a JobRepository. When using the @EnableBatchProcessing annotation a JobRepository is provided for you. Spring batch has features to support our other requirements, like parallel processing, and pausing and restaring job execution based on business requirements. Also, the idea of restarting based on other failures is baked into Spring Batch.

Useful links:

• Spring Batch docs:
• A simple example of using Spring Batch with kafka
• Spring Batch error handling  
• Configuring and running a job in Spring Batch
• Stopping a Spring Batch job for “business reasons”
•  You don’t have to use XML to configure the job
• KafkaConsumer and KafkaItemReader are not thread safe

• This might seem like the simplest approach. The KafkaConsumer has pause and resume methods. 

•  The ConcurrentMessageListenerContainer allows for easy instantiation of multiple concurrent consumers and it allows for easy pause and resume of consumers with a single call to the consumer container.

•  job persistence for free in the form of a JobRepository.
• Spring batch has features to support other requirements, like parallel processing,
•  restarting based on other failures is baked into Spring Batch.

•  it doesn’t have two other requirements which are to save the job and support some form of concurrency
• Note also that KafkaConsumer is not thread safe so using it in concurrent scenarios may not be feasible.

• Implies more devs efforts than in Option A as some codebase rearrangements required

• Note that Spring Batch uses a KafkaItemReader which is based on KafkaConsumer, so it is also not thread safe, so implementing a concurrent scenario with Spring Batch may not be easily done