消息预处理（主题 - 主题） - Kafka Connect API vs. Streams vs Kafka Consumer？

Question

我们需要对从一个主题到另一个主题的每条消息（使用不同的密钥解密/重新加密）进行一些预处理。

我一直在研究使用 Kafka Connect，因为它提供了很多开箱即用的好东西（配置管理、偏移存储、错误处理等）。

但我也觉得我将结束实现 SourceConnector 和 SinkConnector 只是在两个主题之间移动数据，而这两个接口都不打算做 Topic A -> (Connector) -> Topic B。这是正确的方法吗？我应该只使用SinkConnector 并让我的SourceTask.put() 完成所有逻辑来写入Kafka吗？

其他选项是 KafkaConsumer/Producer 和或 Streams，但这些将需要自己的实例来运行逻辑，而不是偏移重试错误处理。

Answer 1

提供了很多开箱即用的好东西（配置管理、偏移存储、错误处理等）

配置管理不应该比重新部署应用程序更难，但这取决于您可能拥有或不拥有的任何版本控制或 CI/CD 管道。

Kafka Producer/Consumer 和 Streams 提供偏移管理，您只需将其配置为执行除默认值之外的任何操作。

错误处理有据可查，如果您关心检测错误，请不要一劳永逸。 Connect 本身将在严重错误条件下停止消费和生产，而不是重试或跳过消息。

这些接口都不打算做Topic A -> (Connector) -> Topic B"

您见过 Confluent Replicator（授权产品）吗？那是两个主题之间的 Kafka Connect。

否则，你见过 MirrorMaker 吗？这是一个生产者-消费者对，通常用于在各个集群之间复制数据，但可以与相同的源和目标设置一起使用。您只需要确保您没有创建反馈循环。您需要对其应用“自定义逻辑”（并更改主题名称），called a Handler class that is placed on your Kafka classpath

bin/kafka-mirror-maker.sh

...

--message.handler <String: A custom      Message handler which will process
  message handler of type                  every record in-between consumer and
  MirrorMakerMessageHandler>               producer.
--message.handler.args <String:          Arguments used by custom message
  Arguments passed to message handler      handler for mirror maker.
  constructor.>

Confluent MirrorMaker documentation
Kafka MirrorMaker documentation

---

没有什么能阻止您实现 Connect API，而且它可能比没有外部集群管理器的 Kafka Streams 应用程序更容易管理。另外，由于 Connect 是一个 Java 库，理论上您可以在其中内部使用 Streams 库。