【发布时间】：2021-04-26 15:06:18
【问题描述】：
我有一个简单的用例：聚合一段时间（1 分钟）内的警报并发送一封电子邮件，其中包含在此期间发生的所有警报。
所以总的来说，我可以毫无问题地定义一个窗口、按用户聚合警报并使用 Kafka Streams 发送电子邮件。如果警报定期出现，它就像一个魅力。如果没有更多警报，将永远不会触发发送电子邮件。原因很明显：Kafka Streams 逻辑是由传入事件触发的。因此，如果主题中没有新事件 - 什么都不会发生。
这是一个显示问题的简单用例：
Time: 0:00:00
-> Event 1
-> Event 2
-> Event 3
Time: 0:01:00
-> Event 4 (This will trigger email with events 1,2,3)
因此，如果没有事件 4 - 将永远不会发送任何电子邮件。
我的一个解决方案是编写自定义处理器并定义将发送电子邮件的调度程序。理想情况下，我仍然会使用官方 Kafka API 进行分组、聚合，并且只使用调度程序逻辑发送电子邮件。在这种情况下，我似乎需要将 kvStore 共享给处理器（java）。不幸的是，我没有找到关于该用例的任何有据可查的文章。
我还尝试将所有内容“委托”到处理器上（将新事件添加到自定义存储中，检查所有事件，发送电子邮件并从存储中删除已发送事件），但看起来并不对我来说干净”。
var kvStore: KeyValueStore<Window, AlertNotificationMessage>? = null
override fun init(context: ProcessorContext) {
kvStore = context.getStateStore(AlertMessageProcessorSupplier.STORE_NAME)
as KeyValueStore<Window, AlertNotificationMessage>
context.schedule(
Duration.ofSeconds(AlertProcessor.AGGREGATION_SECONDS),
PunctuationType.WALL_CLOCK_TIME,
this::sendNotification
)
}
override fun process(key: Windowed<String>, value: AlertNotificationMessage) {
kvStore?.put(
Window(key.key(), key.window().start(), key.window().end()),
value
)
}
private fun sendNotification(callTime: Long) {
kvStore?.all()?.forEach {
if (it.key.endMs < callTime) {
notificationService.send(it.key.recipientId, it.value)
kvStore?.delete(it.key)
}
}
}
有没有人有类似的用例和任何好的解决方案也适用于多分区/故障转移环境？
【问题讨论】：
标签： apache-kafka apache-kafka-streams