[FLINK-30564] Select from a new table with Kafka LogStore crashes with UnknownTopicOrPartitionException

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Component/s: Table Store
• Assignee: Unassigned
• Reporter: Alex Sorokoumov
• Created: 04/Jan/23 23:34
• Updated: 29/Mar/23 01:44
• Resolved: 29/Mar/23 01:44

Description

Selecting from newly created table that uses Kafka as a Log Store creates a job that crash-loops with UnknownTopicOrPartitionException: This server does not host this topic-partition exception. This happens because neither CREATE TABLE nor SELECT FROM create the underlying topic. 

Steps to reproduce:

CREATE TABLE word_count (
    word STRING PRIMARY KEY NOT ENFORCED,
    cnt BIGINT
) WITH (
    'connector' = 'table-store',
    'path' = 's3://my-bucket/table-store',
    'log.system' = 'kafka',
    'kafka.bootstrap.servers' = 'broker:9092',
    'kafka.topic' = 'word_count_log',
    'auto-create' = 'true',
    'log.changelog-mode' = 'all',
    'log.consistency' = 'transactional'
);

SELECT * FROM word_count; 

 

JM logs:

flink          | 2023-01-04 23:27:24,292 ERROR org.apache.flink.runtime.source.coordinator.SourceCoordinatorContext [] - Exception while handling result from async call in SourceCoordinator-Source: word_count[1]. Triggering job failover.
flink          | org.apache.flink.util.FlinkRuntimeException: Failed to list subscribed topic partitions due to
flink          |     at org.apache.flink.table.store.shaded.connector.kafka.source.enumerator.KafkaSourceEnumerator.checkPartitionChanges(KafkaSourceEnumerator.java:234) ~[flink-table-store-dist-0.4-SNAPSHOT.jar:0.4-SNAPSHOT]
flink          |     at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$null$1(ExecutorNotifier.java:83) ~[flink-dist-1.16.0.jar:1.16.0]
flink          |     at org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) [flink-dist-1.16.0.jar:1.16.0]
flink          |     at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [?:1.8.0_352]
flink          |     at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_352]
flink          |     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) [?:1.8.0_352]
flink          |     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) [?:1.8.0_352]
flink          |     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_352]
flink          |     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_352]
flink          |     at java.lang.Thread.run(Thread.java:750) [?:1.8.0_352]
flink          | Caused by: java.lang.RuntimeException: Failed to get metadata for topics [word_count_log].
flink          |     at org.apache.flink.table.store.shaded.connector.kafka.source.enumerator.subscriber.KafkaSubscriberUtils.getTopicMetadata(KafkaSubscriberUtils.java:47) ~[flink-table-store-dist-0.4-SNAPSHOT.jar:0.4-SNAPSHOT]
flink          |     at org.apache.flink.table.store.shaded.connector.kafka.source.enumerator.subscriber.TopicListSubscriber.getSubscribedTopicPartitions(TopicListSubscriber.java:52) ~[flink-table-store-dist-0.4-SNAPSHOT.jar:0.4-SNAPSHOT]
flink          |     at org.apache.flink.table.store.shaded.connector.kafka.source.enumerator.KafkaSourceEnumerator.getSubscribedTopicPartitions(KafkaSourceEnumerator.java:219) ~[flink-table-store-dist-0.4-SNAPSHOT.jar:0.4-SNAPSHOT]
flink          |     at org.apache.flink.runtime.source.coordinator.ExecutorNotifier.lambda$notifyReadyAsync$2(ExecutorNotifier.java:80) ~[flink-dist-1.16.0.jar:1.16.0]
flink          |     ... 7 more
flink          | Caused by: java.util.concurrent.ExecutionException: org.apache.flink.table.store.shaded.org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.