Hi,
This issue was reported earlier in this post , for Kafka 0.9:
https://stackoverflow.com/questions/41177614/kafka-0-9-java-consumer-skipping-offsets-during-application-restart
However, I see the same issue with the Kafka 0.10 as well.... In summary:
1. we start our consumer app, and in the event poll loop we print starting
offset for each partition for this consumer as following:
while
(
true
)
{
boolean
isPollFirstRecord
=
true
;
ConsumerRecords
<
String
,
String
>
records
=
consumer
.
poll
(
pollIntervalMs
);
for
(
ConsumerRecord
<
String
,
String
>
record
:
records
)
{
if
(
isPollFirstRecord
)
{
isPollFirstRecord
=
false
;
logger
.
info
(
"Start offset for partition {} in this poll : {}"
,
record
.
partition
(),
record
.
offset
());
}
// .... do some processing of the event
}
if
(!
records
.
isEmpty
())
{
consumer
.
commitAsync
(
offsetLoggingCallback
);
}
}
2. we commit offsets using the consumer.commitAsync(offsetLoggingCallback)
method, with our OffsetLoggingCallbackImpl implementation class instance (see
above)
3. in the OffsetLoggingCallbackImpl's
onComplete
(
Map
<
TopicPartition
,
OffsetAndMetadata
>
offsets
,
Exception
exception
)
we print committed offsets for each topic/partition
4. we stop our application (consumer) - and note the committed offsets reported
from the OffsetLoggingCallbackImpl class. Let's say we see something like this:
Offset position during the commit for consumerId : consumerThread1, partition :
1, offset : 3400
Offset position during the commit for consumerId : consumerThread1, partition :
2, offset : 3100
5. we check offsets using the OffsetChecker utility:
/opt/kafka/bin/kafka-consumer-groups.sh --new-consumer --describe --group
varnish-s3 --bootstrap-server localhost:9092
ISSUE #1: (not critical) for most of partitions, we see the same committed
offsets as what the OffsetLoggingCallback reported; but for some ( 1 or 2 out
of 40 partitions) - we see the offset reported by Kafka being larger than that
from step 4.
6. we start our application (consumer) again, and check the offsets printed
from the first poll loop:
ISSUE #2: (critical): for most partitions - the offset form Step 6 is the same
as offsets from Step 5. But for some partitions (1 or 2 out of 40, but not the
same as in ISSUE #1) - the starting offsets are AFTER the offsets reported in
Step 5:
Start offset for partition 1 in this poll : 3900
Start offset for partition 2 in this poll : 3600
So, it means that after the restart of our app we have missed events [3400 -
3900] for partition 1, and events [3100-3600] for partition 2.
Here are example results for a couple of partitions:
[kafka_offsets.png]
One interesting point: looks like all differences are 500 events! I wonder if
this is some default buffer size somewhere....
Any idea if it is an issue on the app side or possibly a bug in Kafka?
thanks,
Marina
Sent with [ProtonMail](https://protonmail.com) Secure Email.
