Kafka.consumer.SimpleConsumer: Повторно подключиться из-за ошибки сокета: java.nio.channels.ClosedChannelException

Я использую простой потребитель для kafka, такой как это:

int timeout = 80000;
int bufferSize = 64*1024;
consumer = new SimpleConsumer(host, port,timeout, bufferSize, clientName);

Это работает отлично в течение нескольких часов, но я получаю исключение позже на kafka.consumer.SimpleConsumer: Повторно подключиться из-за ошибки сокета:

java.nio.channels.ClosedChannelException

и потребители останавливаются... кто-нибудь сталкивался с этой проблемой раньше?

Ответы

Ответ 1

Немного другой вопрос, но, возможно, с той же основной причиной и решением был дан ответ, соответствующая часть:

Вы закрыли канал и все еще пытаетесь его использовать.

Есть несколько проблем с вашим кодом.

Во-первых, ваш тест на EOS неисправен. Убрать ограничение() == 0 test. Это не указывает на EOS, это просто указывает на чтение нулевой длины, которое может происходить в неблокирующем режиме в любое время. Это не значит, что пир закрыл свой конец соединения, и это не значит, что вы должны закрыть свой конец.

Во-вторых, закрытие канала также закрывает сокет. Вы должны закрыть только канал, а не сокет.

В-третьих, закрытие канала отменяет ключ. Вам не нужно следовать каждому закрытию с отменой.

Возможно, вам также не удалось проверить, является ли готовый ключ действительным в цикле выбора, прежде чем использовать его, например, для чтения.