Bug Report

We recently discovered that the Asn1EncodedDataRouter was re-processing messages that had been published to topic.Asn1EncoderOutput in our ODE deployment. We've been able to identify the root cause, and reproduce the issue in a new, test consumer subscribed to the same topic.

Symptoms

ODE Logs

Below is an excerpt from the ODE logs (the full 5,000 lines captured can be found in
ode_logs_abbreviated.txt:

[Lines 441-443]

2020-02-10 21:29:13 [pool-3-thread-1] WARN  ConsumerCoordinator - Auto offset commit failed for group AsnCodecRouterServiceController: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records. 
2020-02-10 21:29:14 [pool-3-thread-1] DEBUG MessageConsumer - Asn1EncoderConsumer consuming 406 message(s) 
2020-02-10 21:29:14 [pool-3-thread-1] DEBUG Asn1EncodedDataRouter - Consumed: <?xml version="1.0"?><OdeAsn1Data> ... <streamId>43afff7d-7c47-4941-b89e-9455e52d4687</streamId> ... <odeReceivedAt>2020-02-08T22:09:40.335Z</odeReceivedAt> ... <recordId>BB875239</recordId> ... </OdeAsn1Data> 

...

[Lines 3712-3714]

2020-02-10 21:38:24 [pool-3-thread-1] WARN  ConsumerCoordinator - Auto offset commit failed for group AsnCodecRouterServiceController: Commit cannot be completed since the group has already rebalanced and assigned the partitions to another member. This means that the time between subsequent calls to poll() was longer than the configured max.poll.interval.ms, which typically implies that the poll loop is spending too much time message processing. You can address this either by increasing the session timeout or by reducing the maximum size of batches returned in poll() with max.poll.records. 
2020-02-10 21:38:27 [pool-3-thread-1] DEBUG MessageConsumer - Asn1EncoderConsumer consuming 406 message(s) 
2020-02-10 21:38:27 [pool-3-thread-1] DEBUG Asn1EncodedDataRouter - Consumed: <?xml version="1.0"?><OdeAsn1Data> ... <streamId>43afff7d-7c47-4941-b89e-9455e52d4687</streamId> ... <odeReceivedAt>2020-02-08T22:09:40.335Z</odeReceivedAt> ... <recordId>BB875239</recordId> ... </OdeAsn1Data> 

On line 442, the Asn1EncodedDataRouter received 406 messages after polling the topic. The next time it polls the topic (on line 3713), it again consumed 406 messages (the same messages, in fact). To emphasize the point the the same messages are being re-processed by the Asn1EncodedDataRouter, I've included an abbreviated version of the first message consumed in both events (lines 443 and 3714). In each case, the first message processed has the same streamId, odeReceivedAt and recordId values.

AsnCodecRouterServiceController Group Status

To further illustrate that messages are being re-processed by the Asn1EncodedDataRouter, I queried the AsnCodecRouterServiceController group status, as shown below:

% ./kafka-consumer-groups.sh --bootstrap-server 10.145.9.204:9092 --describe --group AsnCodecRouterServiceController

GROUP                           TOPIC                   PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             CONSUMER-ID                                     HOST            CLIENT-ID
AsnCodecRouterServiceController topic.Asn1DecoderOutput 0          1892835         1892835         0               consumer-1-dacf3c3a-e5e5-42d2-af89-92cb49230b4c /172.18.0.1     consumer-1
AsnCodecRouterServiceController topic.Asn1EncoderOutput 0          4728            46788           42060           consumer-2-0e2fad59-4dc7-436a-9494-5719b6f9f540 /172.18.0.1     consumer-2

The consumer of topic.Asn1EncoderOutput is 42060 messages behind, apparently "stuck".

Root Cause

As indicated in the ODE logs (lines 441 and 3712), the auto offset commit is continually failing. This is because Kafka 0.10.1.0 has a default max.poll.interval.ms property value of 300000 (5 minutes). If the consumer takes longer than 5 minutes between polls(), Kafka will assume the consumer is down. Then when the consumer attempts an auto commit during the next poll(), it will fail. Since the consumer's progress in the topic's commit log isn't being recorded, on the next poll(), the consumer will receive the same messages it had processed previously.

So with the default max.poll.interval.ms value of 5 minutes, and a default max.poll.records value of 500, the Asn1EncodedDataRouter needs to be capable of processing up to 500 messages in a 5 minute period, between polls(). According to the ODE logs, it took the Asn1EncodedDataRouter 9:13 to process 406 messages. Since it's taking > 5 minutes, it's stuck re-processing data.

Possible solutions

The easy solution would be to adjust the max.poll.interval.ms and max.poll.records values accordingly (maybe 600000 and 100, respectively). I would also argue these (and other configuration values) shouldn't be hardcoded in MessageConsumer.java. Rather, they should be set in a configuration file that gets read at runtime.

Another solution that would probably be better long term would be to extract the "deposit to RSUs" logic into it's own module. I think this is the main reason it's taking 9+ minutes to process 406 messages. This would also be preferable, as deposits to the SDW are currently blocked by the logic that deposits messages to RSUs. The "deposit to SDW" logic is already abstracted away from the Asn1EncodedDataRouter, and exists in the jpo-ode-sdw-depositor module. I'd argue the Asn1EncodedDataRouter should behave similarly for RSU deposits. As it does with SDW messages, it could publish messages destined for RSUs to a topic like topic.RSUDepositorInput, then another module (say, jpo-ode-rsu-depositor) handles the details of pushing those messages out to RSUs.