В чем причина получения ProducerFencedException во время ProducerFencedException?

В моем коде инициализации производителя было состояние гонки. Я исправил это, изменив карту Producer на тип ConcurrentHashMap, чтобы обеспечить потокобезопасность.

Caused by: org.apache.kafka.common.errors.ProducerFencedException: Producer attempted an operation with an old epoch. Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker.

Это сообщение об исключении не очень полезно. Я считаю, что пытающийся означает, что у брокера больше нет записи идентификатора транзакции, отправляемой клиентом. Это может быть потому, что:

• Кто-то другой использовал тот же идентификатор транзакции и уже зафиксировал его. По моему опыту, это менее вероятно, если вы не разделяете идентификаторы транзакций между клиентами. Мы гарантируем, что наши идентификаторы уникальны, используя UUID.randomUUID().
• Время ожидания транзакции истекло, и она была удалена автоматизацией брокера.

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

• transaction.max.timeout.ms - свойство маклер, указывающее максимальное количество миллисекунд, пока транзакция не будет прервана и забыта. По умолчанию во многих версиях Kafka кажется, что это 900000 (15 минут). Документация от Кафки говорит:

  The maximum allowed timeout for transactions. If a client’s requested transaction time exceeds this, then the broker will return an error in InitProducerIdRequest. This prevents a client from too large of a timeout, which can stall consumers reading from topics included in the transaction.

• transaction.timeout.ms - свойство производитель-клиент, которое устанавливает время ожидания в миллисекундах при создании транзакции. По умолчанию во многих версиях Kafka кажется, что это 60000 (1 минута). В документации от Кафки говорится:

  The maximum amount of time in ms that the transaction coordinator will wait for a transaction status update from the producer before proactively aborting the ongoing transaction.

Если свойство transaction.timeout.ms, установленное в клиенте, превышает свойство transaction.max.timeout.ms в брокере, производитель немедленно выдаст что-то вроде следующего исключения:

org.apache.kafka.common.KafkaException: Unexpected error in InitProducerIdResponse
The transaction timeout is larger than the maximum value allowed by the broker 
(as configured by transaction.max.timeout.ms).

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

  @Test
  public void SendOffset_TwoProducerDuplicateTrxId_ThrowException() {
    // create two producer with same transactional id
    Producer producer1 = KafkaBuilder.buildProducer(trxId, servers);
    Producer producer2 = KafkaBuilder.buildProducer(trxId, servers);

    offsetMap.put(new TopicPartition(topic, 0), new OffsetAndMetadata(1000));

    // initial and start two transactions
    sendOffsetBegin(producer1);
    sendOffsetBegin(producer2);

    try {
      // when commit first transaction it expected to throw exception
      sendOffsetEnd(producer1);

      // it expects not run here
      Assert.assertTrue(false);
    } catch (Throwable t) {
      // it expects to catch the exception
      Assert.assertTrue(t instanceof ProducerFencedException);
    }
  }

  private void sendOffsetBegin(Producer producer) {
    producer.initTransactions();
    producer.beginTransaction();
    producer.sendOffsetsToTransaction(offsetMap, consumerGroup);
  }

  private void sendOffsetEnd(Producer producer) {
    producer.commitTransaction();
  }

When running multiple instances of the application, transactional.id must be the same on all instances to satisfy fencing zombies when producing records on a listener container thread. However, when producing records using transactions that are not started by a listener container, the prefix has to be different on each instance.

https://docs.spring.io/spring-kafka/reference/html/#transaction-id-prefix