Блог

Я использую Flink 1.12.1 и соответствующий пакет jar-коннекторов Кафки.

 SELECT 'hello';  

…работает правильно, поэтому кластер запущен и работает.

Затем я создал таблицу Кафки:

 CREATE TABLE myusers (  `id` int NOT NULL,  `level` string not null,  `ts` TIMESTAMP(3) METADATA FROM 'timestamp' ) WITH (  'connector' = 'kafka',  'topic' = 'topic',  'properties.bootstrap.servers' = 'kafka:9092',  'properties.group.id' = 'test-group',  'properties.auto.offset.reset' = '',  'scan.startup.mode' = 'earliest-offset',  'format' = 'debezium-json' );  SELECT * from myusers;  

Однако на дисплее результатов данные не извлекаются. Он оставался пустым и освежал каждую секунду. Через 5 минут истекло время ожидания задания, так как данные не передаются. Я вижу 16 подзадач, все SCHEDULED затем FAILED / CANCELLED через 5 минут.

Я проверил, что могу подключиться к кластеру Кафки и использовать kafka-console-consumer.sh его для перечисления всех сообщений. Я также пытался использовать API потока данных для прямого подключения к той же теме. Работа прошла нормально, и я мог видеть, что необработанные данные печатаются для вывода из строя в этой работе.

Ни в одном из журналов taskmanager, stdout или других журналов не было журналов соединителя Кафки. Я также изменил все уровни журналов на DEBUG и могу видеть много DEBUG журналов, но все еще нет журналов Кафки. С теми же настройками я мог видеть журналы соединителя Кафки при использовании API потока данных.

Я исчерпал возможности устранения неполадок. Итак, мои вопросы:

• Почему при использовании клиента Flink SQL не было никаких журналов соединителя Кафки?
• Есть ли способ увидеть фактическое задание, отправленное в Flink клиентом SQL, чтобы я мог проанализировать, что пошло не так?
• В чем может быть проблема?