Fink: Потеря данных KafkaProducer

Question

Наш потоковый поток Flink публикует сообщения в Kafka. Механизм «повторения» KafkaProducer не срабатывает до тех пор, пока в его внутренний буфер не добавится сообщение.

Если перед этим существует исключение, KafkaProducer выкинет это исключение и похоже, что Flink не справляется с этим. В этом случае будет потеря данных.

Похожие Flink код (FlinkKafkaProducerBase):

if (logFailuresOnly) { 
      callback = new Callback() { 
       @Override 
       public void onCompletion(RecordMetadata metadata, Exception e) { 
        if (e != null) { 
         LOG.error("Error while sending record to Kafka: " + e.getMessage(), e); 
        } 
        acknowledgeMessage(); 
       } 
      }; 
     } 
     else { 
      callback = new Callback() { 
       @Override 
       public void onCompletion(RecordMetadata metadata, Exception exception) { 
        if (exception != null && asyncException == null) { 
         asyncException = exception; 
        } 
        acknowledgeMessage(); 
       } 
      }; 
     } 

Вот сценарий-х мы определили, что приведет к потере данных:

1. Все Кафка брокеры вниз.

   В этом случае перед добавлением сообщения в его буфер KafkaProducer пытается получить метаданные. Если KafkaProducer не может получить метаданные в настроенном тайм-ауте, он выдает исключение.

2. -Память записи не доступен для записи (существующий ошибка в библиотеке KAFKA 0.9.0.1)

https://issues.apache.org/jira/browse/KAFKA-3594

В обоих вышеуказанных случаях, KafkaProducer не будет повторять, и Флинка будет игнорировать сообщения , сообщения даже не регистрируются. Исключение составляет, но не сообщения, которые не удались.

Возможные обходные пути (настройки Кафка):

1. Очень высокое значение тайм-аута метаданных (metadata.fetch.timeout.ms)
2. Очень высокое значение для буфера истечения срока действия (request.timeout.ms)

Мы по-прежнему изучаем возможные побочные эффекты изменения вышеуказанных настроек кафки.

Итак, наше понимание правильное? Или можно ли избежать этой потери данных, изменив некоторые настройки Flink?

Спасибо.

Answer 1

Вот что я думаю о ваших вопросах. Смотрите один из Кафки гарантирует первое:

Для темы с фактором репликации N, мы будем терпеть до сбоев сервера N-1 без потери каких-либо записей , совершенных в журнал.

Во-первых, он заботится о сообщениях или записях, зафиксированных в журнале. Любые записи, которые не были доставлены, не считаются совершенными. Во-вторых, если все ваши брокеры были опущены, будет некоторая потеря данных.

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

• block.on.buffer.полный = истинное
• ACKs = все
• повторы = max_value
• max.in.flight.requests.per.connection = 1
• Использование KafkaProducer.send (запись, обратный вызов) вместо отправки (запись)
• unclean.leader.election.enable = ложь
• replication.factor> min.insync.replicas
• min.insync.replicas> 1