我想从kafka主题中定期读取一批消息，或者当读取的消息数达到一定数量时，将它们作为批处理发送到下游系统 . 目前，我的kafka拓扑结构由处理器终止，处理器保存消息，然后使用punctuate方法逐步处理批处理 .
但是，我不确定这是完美的，因为如果应用程序在调用punctuate方法之前崩溃，我认为一些消息会丢失（即消费者认为它已经完成了它们但它们不会出现在下游系统中） .
batchQueue = new LinkedBlockingQueue<String>(batchSize);
KStream<String, String> inputStream = builder
.stream(Serdes.String(), Serdes.String(), "source-topic")
.process(new ProcessorSupplier<String, String>() {
@Override
public Processor<String, String> get() {
return new AbstractProcessor<String, Wrapper>() {
@Override
public void init(ProcessorContext context) {
super.init(context);
context.schedule(flushPeriod);
}
@Override
public void process(String key, String value) {
batchQueue.add(value);
if (batchQueue.remainingCapacity() == 0) {
processQueue();
}
}
@Override
public void punctuate(long timestamp) {
processQueue();
context().commit();
}
}
@Override
public void close() {}
};
}
});
-
有没有办法让这种方法更健壮？也许是窗口，但我真的不明白这一点 .
-
我是否必须使用Kafka连接 . 我'm leaning away from this due to it'的错误处理能力：https://groups.google.com/forum/#!topic/confluent-platform/OBuLbVHbuyI
1 回答
我建议将数据转换部分（我将使用Kafka的Streams API）和数据提取部分（您写入下游系统（我为其使用Kafka的Connect API））分离 .
简而言之，为什么要将转换逻辑与之相结合，并且需要担心这个数据最终会转发到哪个下游系统的细节（这里是：昂贵的插入！）？理想情况下，转型的责任应该只是转型，而不应该关注外部下游系统的运营方面 . 例如，如果您最终想要将转换后的数据转发到第二个下游系统（或第三个......），那么耦合方法意味着您必须更新/重新部署/ ...您的应用程序，甚至虽然它的转换逻辑都没有改变 .
解耦转换和提取的另一个好处是，您的转换逻辑将更加简单，因为它不必考虑由于下游系统缓慢，不可用等导致的故障 . 例如，它不需要实现/测试复杂的重试逻辑 .
不，你不需要使用Kafka Connect，但它可以说是完成这项任务的最佳工具 .
在最新版本的Kafka Connect中，错误处理实际上非常好 . 此外，通过为Connect提供更强大的转换器（思考：串行器/解串器），可以轻松解决链接讨论中的问题 .
此外，正如该链接中所提到的，当您在将数据写入Kafka之前验证数据的兼容性时，在那里讨论的特定问题变得更少 . 您可以通过利用Confluent的架构注册表（https://github.com/confluentinc/schema-registry，docs或类似工具）来实现这一点 . 由于您提出了问题"how can I make this more robust"，考虑data serialization and evolution是我在部署到 生产环境 之前要考虑的另一个重要方面 .
希望这可以帮助！