Kafka Streams join stops processing after a few days

We have a Kafka Streams topology involving an inner join between two streams that are derived from the same topic. This works fine on new data as it comes in and when reprocessing smallish samples. The problem is that if we reprocess the entire input topic (21M messages), processing stops in the joining service after a few days, resulting in far fewer output messages than expected.

Some ASCII art to illustrate the topology:

Input -> A --------- C -> output
          \--- B ---/

Given that A produces a message A1, this is consumed by B which produces a message B1. When C gets message B1, it joins it to A1 and processes the two to produce a new message C1.

Some extra info that might be relevant:

• A is very fast - it can process 70K input messages a second
• B and C are very slow - they each process about 10 messages a second
• Corresponding messages on A and B have the same timestamp and the same key, and the topics have the same number of partitions
• It’s a KStream-KStream inner join with a time window of 10 seconds
• All three are Scala components using org.apache.kafka.kafka-streams-scala 2.6.0, configured with processing-guarantee = "at_least_once" and auto-offset-reset-config = "earliest"
• This a three-node MSK cluster (Kafka 2.2.1); the services run on EKS
• No errors occur in any of the components

When the problem has occurred, we observe that C’s lag on A is 0 - ie C has consumed all the old messages from A, even though B is still producing new messages based on A messages that should trigger joins. At that point, all the messages from A are present on C’s underlying temporary join topic.

Once B has caught up, new messages that appear on A get processed immediately by B and then C. Since processing resumes when more messages appear, it seems that C hasn’t crashed, it just thinks it has no work to do because it’s prematurely reached the end of A’s topic.

Mitigation we’ve tried:

• To avoid retention-related problems, all the topics, including C’s internal KSTREAM-JOIN* topics, are set to compact with infinite retention (retention.bytes and retention.ms set to -1)
• We originally joined A to B, but actually, C is only interested if B has produced a message so now we join B to A. The behaviour is the same in both cases.

The simplest solution might be to include in B all the data C needs from A instead of doing a join. But as we use Kafka Streams a great deal, we’d like to understand what’s going on here and how to configure it better. I wondered if a larger time window would help but my understanding is that the window is based on the message timestamp, and the timestamps on the corresponding messages in all the topics are all identical.

Does this sound like a familiar problem anyone can shed a light on?