说明：
我有一个kafka流应用程序，它正在使用一个主题。这些活动的数量正在激增。
kafkastream将这些事件作为终端操作使用，并将这些事件合并成一堆（比如1000个事件）并将其写入awss3。
我有一些线程在使用了kafka主题中的事件之后，正在并行地写入s3。
由于某些业务应用程序逻辑和处理，未使用kafka-connector-s3。
问题：
我希望应用程序是容错的，不希望丢失消息。
-->崩溃场景
假设应用程序有10个线程，所有线程都在运行并试图将事件放入s3，在这种情况下，会发生崩溃，因为kafkastream有（enable.auto.commit=false），我们无法手动提交偏移量，并且所有线程都使用了来自kafka主题的消息。在本例中，kafkastreams已经在读取之后提交了偏移量，但它不可能将事件处理到s3。
我需要一个机制，这样我就可以确定在事件成功写入s3文件之前的最后一个偏移量是多少。在崩溃场景中，我应该如何处理这个问题，以及如何管理kafka流中的kafka偏移量，因为我正在使用10个线程。如果有些未能写入s3，而有些通过了呢。如何确保偏移量的排序成功处理到s3？
如果我不清楚如何描述我的问题陈述，请告诉我。
谢谢！
1条答案
按热度按时间rsl1atfo1#
我可以向你保证
enable.auto.commit设置为false在Kafka河。javadocs在https://kafka.apache.org/26/javadoc/org/apache/kafka/streams/streamsconfig.html 状态“enable.auto.commit”（false）-streams客户端将始终禁用/关闭自动提交
你说得对，kafka流会或多或少地定期自动提交。但是，kafka流在提交相应的偏移量之前会一直等待，直到处理完记录。这意味着你至少会得到一次保证，而且不会丢失消息。
据我所知，在将记录发送到s3之前，终端处理器不会阻塞您的应用程序。这意味着，Kafka流无法知道何时发送完成。kafka streams只看到终端处理器完成了处理，然后——如果提交间隔已过——提交偏移量。
你说呢
由于某些业务应用程序逻辑和处理，未使用kafka-connector-s3。
您是否可以将业务应用程序逻辑放入kafka streams应用程序中，将结果写入带有运算符的kafka主题
to()，然后使用kafka-connector-s3将该主题中的消息发送到s3？我不是一个连接Maven，但我想这将确保消息不会丢失，并将使您的实现更简单。