Context

I have a Spring Batch project acquiring messages from Kafka topics using KafkaItemReader, every 4-hours scheduling, for a total of hundred-thousands messages per day. Unfortunately, I don't have enough expertise with Kafka to be independent on the investigation.

I will give as much detail as possible, and will gladly edit the post to answer to request for additional details in comments or answers. I simply don't know exactly where to look for in order to conduct a proper investigation.

Sympthoms

We have detected that a number of messages is being acquired duplicate from the production system. Running specific queries using the business key reports that about thousands messages per day are being acquired twice.

We do not have the unique message key , both because Spring Batch's KafkaItemReader returns only the message body, and because I discovered that the producer won't create a message key at all (at least from Kafka Magic i see it blank, see example below)

Code structure, partitioning of the topics

We are trying to leverage on the best features of Spring Batch. The job that imports Kafka messages is Spring-partitioned into partitions each referring to a separate topic. All Kafka partitions are configured in the same KafkaItemReader (see code), then an idiomatic reader-processor-writer step is executed.

- root job
  - partition step by companyId (each maps unique Kafka topic) + AsyncTaskExecutor
    - chunked step
      - KafkaItemReader
      - CompositeItemProcessor
      - Custom writer

Here is the code that instantiates the KafkaReader into each partitioned step

    @StepScope
    @Bean
    protected  KafkaItemReader kafkaItemReader(
            @Value("#{stepExecutionContext[T(Constants).KEY_COMPANY_CODE]}") int companyCode,
            @Value("#{stepExecutionContext[T(Constants).KEY_TOPIC_NAME]}") String topicName,
            KafkaProperties kafkaProperties,
            MyConfigurationProperties myConfigurationProperties
    ) {
        var partitions = myConfigurationProperties.getTransitTopicInPartitions(); //Integer[], simplified syntax
        var kafkaPropertiesMap = kafkaProperties.buildConsumerProperties();
        return new KafkaItemReaderBuilder()
                .name("kafkaItemReader_" + companyCode)
                .topic(topicName)
                .consumerProperties(toProperties(kafkaPropertiesMap))
                .partitions(partitions)
                .partitionOffsets(new HashMap<>())
                .build();
    }

Let's focus on here. All Kafka partitions are mapped to the same reader. In the future we may think of refactoring the whole process partitioning so that each thread-scoped step reads from a single partition, enhancing parallelism, but that's not today. I know I have used partitionOffsets(new HashMap<>()), which seems a bit counterintuitive, but I don't know if it is necessary to use some kind of persistent object in between Spring Batch runs.

In other words: that line seems to me a major suspect because I don't have a real understanding of its functioning (and is mandatory otherwise you get NPE). But the docs say Passing an empty map makes the reader start from the offset stored in Kafka for the consumer group ID. , which is what I want.

Here are some Kafka properties:

SPRING_KAFKA_BOOTSTRAPSERVERS=zzz
SPRING_KAFKA_CONSUMER_AUTOOFFSETRESET=earliest
SPRING_KAFKA_CONSUMER_GROUPID=xxx
SPRING_KAFKA_PROPERTIES_BASIC_AUTH_CREDENTIALS_SOURCE=USER_INFO
SPRING_KAFKA_PROPERTIES_SASL_MECHANISM=SCRAM-SHA-512
SPRING_KAFKA_PROPERTIES_SCHEMA_REGISTRY_URL=xxx
SPRING_KAFKA_PROPERTIES_SECURITY_PROTOCOL=SASL_SSL

And for the specific reader, our application.yaml defines partitions as 0,1,2,3,4,5,6,7,8,9 like they are configured in PRD.

No message duplication in Kafka (at least...)

Customer team said that they did some checks on the producing application and Kafka side, and found no duplication. I myself ran a search via KafkaMagic for a specific business key taken as sample from database, and I found a single message. I didn't execute other sample-based investigations.

Wrong number of partitions acquired on day 0

Something I would like to share with audience. We went live on May 31st, but for a misunderstanding (e.g. "no one told me I had to configure 10 partitions"), only partitions 0,1,2,3,4 were active in production. When it was found (a week later) that we didn't have enough records at database and that the partitions were, in reality, 10, I changed the yaml and requested a production deployment without changing the group id so that old partitions 0-4 could resume at their offset and partitions 5-9 start from scratch having no offset.

We don't have any record telling whether duplications existed before this change or not, but the customer team suspects that "since the day we started acquiring all partitions, we may have introduced duplication", without further argumenting it.

The question

Given this setup, do you see anything wrong that could blatantly cause message duplication while acquiring?

What can I do in order to investigate the root cause of this message duplication? Given the number of messages, I don't have the abiity to go by Java debug.

Do you have insights?