Kafka Batching

GitHub Copilot — Kafka Streams Rebalancing and Retry Issue: Root Cause Analysis

Mode: ASK ONLY — STRICTLY NO CODE CHANGES

I am working on Jira PRGCRM-5120, migrating the OneCRM DataService Kafka consumers from Spring @KafkaListener to native Kafka Streams (KStream), while preserving the existing batching and business-processing behavior.

The implementation has been deployed in the CAE environment, and I have also performed local testing against the Confluent non-production cluster.

I am experiencing prolonged consumer-group rebalancing in local testing, particularly for OCP CustExp, along with downstream processing failures and retry-related exceptions.

I need a detailed, evidence-based root cause analysis of the actual current source code before making any modifications.

1. Inspect the complete implementation

Read and trace these classes and relevant supporting code:

• ConfluentKafkaStreamsConfig.java

• KafkaStreamsBatchingProcessor.java

• ConfluentKafkaMessageConsumer.java

• ConfluentKafkaStreamsRunner.java

• OCPRecordProcessor.java

• RIRecordProcessor.java

• Relevant downstream service/client implementations, HTTP timeout and retry configurations

• application.yml, application-cae.yml, application-local.yml

• Relevant Kafka Streams and consumer configurations

• Relevant existing unit and integration tests

Do not limit the analysis to individual methods. Trace the complete execution flow and interactions between classes.

2. Investigate the consumer-group rebalancing issue

Observed behavior:

• CAE C1 consumer groups are generally stable and processing messages.

• During local D1 testing, the CustExp group has remained in Preparing Rebalance for approximately 5–7 minutes or longer.

• CustReco and RI CAD behave differently and may remain stable.

• Kafka logs contain:

  • Request joining group due to: group is already rebalancing

  • Changes in consumer-group member counts

  • Failed to lock the state directory for task

• An earlier test also produced InconsistentGroupProtocolException when incompatible consumer protocols shared a group.

Analyze:

1. Could the current code block Kafka Streams processing threads for too long?

2. Is downstream processing executed synchronously inside the Streams thread?

3. Could excessive processing time violate max.poll.interval.ms?

4. Where is max.poll.interval.ms configured, and does that property actually reach the Kafka Streams main consumer?

5. Are consumer group IDs/application IDs configured correctly and uniquely for the three independent Kafka Streams applications?

6. Could multiple instances, duplicate applications, stream-thread configuration or stale group members cause repeated rebalancing?

7. Are Kafka Streams task migration and RocksDB state-directory locks contributing to the problem?

8. Is the original Spring Kafka listener completely disabled so that no incompatible consumer is joining the same group?

9. Could the current topology configuration or bean lifecycle cause additional problems?

Important: Distinguish normal Kafka rebalance messages from actual errors. Do not treat Preparing Rebalance, a member-count difference or a temporary state-directory lock as definitive proof of a code defect.

3. Analyze downstream processing and retry behavior

A previous local log showed:

• DATA_SERVICE_ERROR_RUNTIME_EXCEPTION_RECORD_PROCESSED_OCP

• RetryExhaustedException: Retries exhausted

• An OCP CustExp processing operation lasting approximately 320 seconds

• Failures involving downstream HTTP/Reactor processing

Trace the complete execution path from:

KStream → KafkaStreamsBatchingProcessor.process() → flushInternal() → processCollectedBatch() → batchConsumer.apply() → ConfluentKafkaMessageConsumer.receiveOCPFromConfluent() → OCPRecordProcessor.processOCPCustExpRecords() → downstream client.

Check:

1. Which exact downstream call can take several minutes?

2. What timeout and retry policies are configured?

3. Can retry/backoff settings cause Kafka Streams threads to remain blocked?

4. Are exceptions propagated correctly from downstream processing?

5. Can a downstream failure be caught and logged internally but still result in true being returned to the batching processor?

6. Can unsuccessful batches be retried repeatedly without adequate delay?

7. Could repeated processing of retained batches contribute to high processing time or rebalancing?

8. Is the implementation handling partial batch success correctly?

9. Could repeated retries cause duplicate downstream writes or events?

Identify actual code evidence for each finding. Do not assume the downstream service is unavailable without supporting evidence.

4. Review the custom batching implementation

Inspect KafkaStreamsBatchingProcessor.java, including:

• init()

• process()

• flushInternal()

• collectPendingBatch()

• parseStateStoreEntry()

• processCollectedBatch()

• pendingRecordsCount()

• close()

Verify:

1. Whether batching works correctly for 100 records.

2. Whether the 1000 ms wall-clock punctuation is configured and executed correctly.

3. Whether a slow downstream call blocks subsequent processing or punctuation.

4. Whether pendingRecordsCount() causes unnecessary RocksDB iterations or performance overhead.

5. Whether stateStore.put() and stateStore.delete() are handled correctly.

6. Whether batches are deleted only when processing genuinely succeeds.

7. Whether failed batches can be retried safely.

8. Whether shutdown and rebalancing can result in repeated processing.

9. Whether record metadata, partition, offset and serialization/deserialization remain correct.

10. Whether there are risks of data loss or duplicate processing.

Also verify the implications of using:

Stores.persistentKeyValueStore(...) with .withLoggingDisabled().

Explain the impact on pending batch recovery after an application crash, pod restart or task migration. Do not assume that a persistent local RocksDB store guarantees recovery when its changelog is disabled.

5. Review Kafka Streams configuration

Examine buildStreamsPropertiesMap() and all three StreamsBuilderFactoryBean configurations.

Verify:

• application.id

• num.stream.threads

• auto.offset.reset

• max.poll.interval.ms

• max.poll.records

• commit.interval.ms

• Consumer property inheritance

• Any application-instance or group-ID conflicts

• Whether different streams share any unintended local state directories

• Kafka Streams state restoration and offset behavior

Explain why CAE group membership previously showed 18 members and subsequently showed 6 members. Determine what is confirmed by the configuration and what would require runtime instance information.

6. Identify unnecessary or risky code

Inspect the implementation for:

• Unused classes or methods

• Deprecated wrappers that are no longer called

• Redundant configuration or duplicated logic

• Unnecessary state-store iterations

• Retry logic that may introduce operational instability

• Incorrect exception handling

• Inefficient synchronous operations inside Kafka Streams threads

Do not suggest removing persistent batching or rewriting the overall implementation unless you demonstrate a concrete technical reason.

7. Required output format

Provide the analysis in the following order.

A. Root cause summary

Explain the top 3 most likely causes of prolonged rebalancing, ranked by likelihood.

B. Code evidence

For every finding include:

• Exact file path

• Method name

• Approximate line number

• Relevant code snippet

• Why the code may cause the observed issue

• Whether the finding is CONFIRMED, LIKELY, POSSIBLE, or NOT YET VERIFIED

C. Rebalancing vs downstream failure

Explain whether these are one connected problem or two independent problems, based on available evidence.

D. Recommended minimal fixes

Recommend the smallest safe changes required, with:

• Exact class/method

• What should change

• Why it should change

• Expected benefit

• Potential side effects

• Required tests

Clearly separate code fixes from configuration fixes and infrastructure/downstream fixes.

E. Validation plan

Explain which specific runtime logs and Kafka consumer-group metrics are needed to confirm the root cause.

F. Final recommendation

Identify the single highest-priority issue to investigate or fix first, explaining why.

Strict instructions

• ASK MODE ONLY.

• DO NOT modify, create or delete any files.

• DO NOT automatically generate patches.

• DO NOT refactor the implementation.

• DO NOT modify Kafka consumer groups, offsets, cluster resources or topics.

• DO NOT change shared CAE configuration.

• DO NOT recommend arbitrary timeout increases as a substitute for root cause analysis.

• Do not claim runtime facts that cannot be verified from source code or supplied logs.

• Preserve existing business behavior, batching requirements and consumer-group identities.

• Focus on evidence-based diagnosis and minimal changes.

Start by examining the actual codebase and explain precisely where the rebalancing and retry problems could originate.