Прерывается при присоединении к ioThread / Error при удалении оператора потока в приложении flink - PullRequest
Новая
       42

Прерывается при присоединении к ioThread / Error при удалении оператора потока в приложении flink

0 голосов
/ 29 января 2019

У меня есть потоковое приложение на основе Flink, которое использует источники и приемники apache kafka.Так как в некоторые дни я получаю исключения в случайное время в процессе разработки, и я понятия не имею, откуда они берутся.

Я запускаю приложение в IntelliJ, используя класс mainRunner, и якормить его сообщениями через кафку.Иногда первое сообщение вызовет ошибки, иногда это происходит только после нескольких сообщений.

Вот как это выглядит: 

16:31:01.935 ERROR o.a.k.c.producer.KafkaProducer      - Interrupted while joining ioThread
java.lang.InterruptedException: null
    at java.lang.Object.wait(Native Method) ~[na:1.8.0_51]
    at java.lang.Thread.join(Thread.java:1253) [na:1.8.0_51]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1031) [kafka-clients-0.11.0.2.jar:na]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1010) [kafka-clients-0.11.0.2.jar:na]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:989) [kafka-clients-0.11.0.2.jar:na]
    at org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer.close(FlinkKafkaProducer.java:168) [flink-connector-kafka-0.11_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:662) [flink-connector-kafka-0.11_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43) [flink-core-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117) [flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477) [flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378) [flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) [flink-runtime_2.11-1.6.1.jar:1.6.1]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_51]
16:31:01.936 ERROR o.a.f.s.runtime.tasks.StreamTask    - Error during disposal of stream operator.
org.apache.kafka.common.KafkaException: Failed to close kafka producer
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1062) ~[kafka-clients-0.11.0.2.jar:na]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1010) ~[kafka-clients-0.11.0.2.jar:na]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:989) ~[kafka-clients-0.11.0.2.jar:na]
    at org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProducer.close(FlinkKafkaProducer.java:168) ~[flink-connector-kafka-0.11_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.close(FlinkKafkaProducer011.java:662) ~[flink-connector-kafka-0.11_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.api.common.functions.util.FunctionUtils.closeFunction(FunctionUtils.java:43) ~[flink-core-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.dispose(AbstractUdfStreamOperator.java:117) ~[flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.disposeAllOperators(StreamTask.java:477) [flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:378) [flink-streaming-java_2.11-1.6.1.jar:1.6.1]
    at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711) [flink-runtime_2.11-1.6.1.jar:1.6.1]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_51]
Caused by: java.lang.InterruptedException: null
    at java.lang.Object.wait(Native Method) ~[na:1.8.0_51]
    at java.lang.Thread.join(Thread.java:1253) [na:1.8.0_51]
    at org.apache.kafka.clients.producer.KafkaProducer.close(KafkaProducer.java:1031) ~[kafka-clients-0.11.0.2.jar:na]
    ... 10 common frames omitted
16:31:01.938 ERROR o.a.k.c.producer.KafkaProducer      - Interrupted while joining ioThread

Я получаю около 10-20 из них, и затем кажется, что Flink восстанавливает приложение, и оно снова становится пригодным для использования, и я могу успешнообрабатывать сообщения.

Что может вызвать это?Или как я могу анализировать дальше, чтобы отследить это?

Я использую Flink версии 1.6.1 с Scala 2.11 на Mac с IntelliJ версии 2018.3.2.

1 Ответ

0 голосов
/ 30 января 2019

Я смог решить это.Оказалось, что один из моих потоковых операторов (map-function) выдавал исключение из-за какого-то недопустимого индекса массива.

Невозможно было увидеть это в журналах, только когда я пошагово разворачивалРазобрав приложение на более мелкие части, я, наконец, получил именно это исключение в журналах, и после исправления очевидной ошибки в доступе к массиву вышеупомянутые исключения (java.lang.InterruptedException и org.apache.kafka.common.KafkaException) исчезли.

...