在Kafka Streams应用程序中启动新线程（使用编程方式）是否可取？

Question

我们正在开发一个使用低级别处理器API的Kafka流应用程序。

根据Kafka的文档，所有线程和并行性都由流线程和流任务处理。使用主题上的分区，并行性也是可伸缩的。

public class Processor implements Processor<K, V> {

@Override
  public void process(String key, V value) {

      //Do processing on the stream thread itself
      ...

      // Write back to output topic
      context.forward(key, updatedValue)
    }); 
  }
}

public class Processor implements Processor<K, V> {

@Override
  public void process(String key, V value) {

  //Spawn new thread to do the processing
    ExecutorService executor = Executors.newSingleThreadExecutor();
    executor.submit(() -> {
      String threadName = Thread.currentThread().getName();
      System.out.println("Hello " + threadName);

      //Do more processing
      ...

      // Write back to output topic
      context.forward(key, updatedValue)
    }); 
  }
}

我尝试了最基本的代码，但不能确定它是否介入了Kafka提供的自动功能。例如自动提交偏移量、超时等。

还是坚持Kafka streams已经提供的默认行为并利用stream线程快速处理数据总是更好？

Answer 1

不建议启动您自己的线程，因为这破坏了Kafka Streams的容错保证。如果process()返回，则Kafka Streams假设消息已被完全处理，并且所有潜在的输出消息都通过forward()发送。在这种情况下，Kafka流可能提交输入记录偏移量。

但是，如果您在后台线程中处理消息，而线程处理失败，则Kafka流将不知道它的任何信息，因此，即使发生故障并且消息丢失，也可能提交偏移量。

此外，后台线程在返回process()后不允许调用forward()。如果forward()被称为process()的“外部”，则Kafka流将引发异常。

使用您自己的后台线程并保留至少一次的处理保证并不是不可能的，但是，这是相当复杂的，因此不推荐使用。