Alex Sorokoumov created FLINK-30564:
---------------------------------------
Summary: Select from a new table with Kafka LogStore crashes with
UnknownTopicOrPartitionException
Key: FLINK-30564
URL: https://issues.apache.org/jira/browse/FLINK-30564
Project: Flink
Issue Type: Bug
Components: Table Store
Reporter: Alex Sorokoumov
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:
{noformat}
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; {noformat}
JM logs:
{noformat}
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. {noformat}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
