Fix partitions initial offsets in Kafka connector

[anestoruk]

Hello,

I've discovered a bug in Kafka initial offsets functionality that was introduced some time ago. Steps to reproduce this bug:

1. Start Kafka source Jet job with initial offsets pointing to the last offset for each partition for given topic.*
2. Suspend the job before any new records are added to the Kafka topic and then consumed by the Jet job (consuming at least 1 record from each partition will result in proper behavior, because internally stored offsets array will be updated to correct values).
3. Resume the job. After doing it, Kafka processor will set consumer positions to the previous offsets incremented by 1, this will exceed the size of the partition and will result in consumer reading records from beginning of each partition.

*To be precise, it's not necessary to provide initial offsets pointing specifically to the end of each partition, because in theory any initial offset could be used and the wrong behavior would occur (i.e. after the job is resumed, the processor will add +1 to the initial offsets), but in reality using any values other than the "last offset" of a given partition will (in most cases) cause the Jet job to consume some records before we have a chance to suspend the job.

I made a fix for this problem by decrementing the value that is put into the StreamKafkaP's internal offsets map at the time when seekToInitialOffsets() is being executed.

Checklist:

• Labels (Team:, Type:, Source:, Module:) and Milestone set
• Label Add to Release Notes or Not Release Notes content set
• Request reviewers if possible
• Send backports/forwardports if fix needs to be applied to past/future releases
• New public APIs have @Nonnull/@Nullable annotations
• New public APIs have @since tags in Javadoc

[devOpsHazelcast]

Can one of the admins verify this patch?

[devOpsHazelcast]

Can one of the admins verify this patch?

[devOpsHazelcast]

Can one of the admins verify this patch?

[devOpsHazelcast]

Can one of the admins verify this patch?

[TomaszGaweda]

run-lab-run