Major Description
We have a microservices that use Kafka Streams which stuck in initialization of stream topolgy while filling StateStore from Kafka using KafkaConsumer. Microservice is build with Kafka Streams 0.10.2.1-cp1 (Confluent 3.2.1) but environment runs Kafka cluster 1.0.0 (Confluent 4.0.0).
We reproduced this problem several times by restarting microservices and eventually had to reset the stream offsets to beginning in order unblock microservices.
While investigating this problem more deeply we found out that StateStore (0.10.2.1) stuck in loading data using ProcessorStateManager. It uses KafkaConsumer (0.10.2.1) to fill the store and it calculates offsets like this:
Fetcher:524
long nextOffset = partRecords.get(partRecords.size() - 1).offset() + 1;
Get the latest offset from records (which were got from poll) plus 1.
So the next offset is estimated.
In Kafka 1.0.0 otherwise broker explicitly provides the next offset to fetch:
long nextOffset = partitionRecords.nextFetchOffset;
It returns the actual next offset but not estimated.
That said, we had a situation when StateStore (0.10.2.1) stuck in loading data. The reason was in ProcessorStateManager.restoreActiveState:245 which kept spinning in consumer loop 'cause the following condition never happened:
} else if (restoreConsumer.position(storePartition) == endOffset) { break; }
We assume that consumer 0.10.2.1 estimates endoffset incorrectly in a case of compaction.
Or there is inconsistency between offsets calculation between 0.10.2.1 and 1.0.0 which doesn't allow to use 0.10.2.1 consumer with 1.0.0 broker.