[
https://issues.apache.org/jira/browse/KAFKA-10454?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17188334#comment-17188334
]
Levani Kokhreidze commented on KAFKA-10454:
-------------------------------------------
Workaround we have used so far is to force repartitioning with num of
partitions that equal to table source topic. Based on example from the
integration test posted above, it would look like this:
{code:java}
/stream
.selectKey((key, value) -> key)
.repartition(Repartitioned.numberOfPartitions(2))
.join(table, (value1, value2) -> value2)
.to(outputTopic);
{code}
> Kafka Streams Stuck in infinite REBALANCING loop when stream <> table join
> partitions don't match
> -------------------------------------------------------------------------------------------------
>
> Key: KAFKA-10454
> URL: https://issues.apache.org/jira/browse/KAFKA-10454
> Project: Kafka
> Issue Type: Bug
> Components: streams
> Affects Versions: 2.6.0
> Reporter: Levani Kokhreidze
> Priority: Major
>
> Here's integration test: [https://github.com/apache/kafka/pull/9237]
>
> From the first glance, issue is that when one joins stream to table, and
> table source topic doesn't have same number of partitions as stream topic,
> `StateChangelogReader` tries to recover state from changelog (which in this
> case is the same as source topic) for table from partitions that don't exist.
> Logs are spammed with:
>
> {code:java}
> [2020-09-01 12:33:07,508] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,508] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,508] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,510] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,510] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,510] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,510] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,513] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,513] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,513] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,513] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,515] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,515] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,515] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,515] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,517] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,518] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,518] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,518] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,520] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,520] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,520] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,520] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,522] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,522] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,522] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,522] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,524] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,524] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,525] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,525] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,526] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,527] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,527] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,527] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-3 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,529] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-0 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,529] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-1 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> [2020-09-01 12:33:07,529] INFO stream-thread
> [app-StreamTableJoinInfiniteLoopIntegrationTestloop-86ae06c3-5758-429f-9d29-94ed516db126-StreamThread-1]
> End offset for changelog
> topic-b-StreamTableJoinInfiniteLoopIntegrationTestloop-2 cannot be found;
> will retry in the next time.
> (org.apache.kafka.streams.processor.internals.StoreChangelogReader:716)
> {code}
> And Kafka Streams never moves to RUNNING state.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
