apache-kafka - 如何对齐（延迟处理）来自两个主题的 Kafka 事件（通过内部属性）？或者如何顺序处理来自两个主题的事件？

Question

为什么我需要这个：

我正在实施一个系统来安排集群上的虚拟机。虚拟机从集群请求资源，我有责任将给定的 RAM 和 CPU 调度到一个且只有一个虚拟机。我想保证这一点的唯一方法是一个一个地处理请求。

创建 VM 的请求会发布到requests主题（在时间轴上方）。集群状态（已用/总资源cluster）作为一系列更新存储在主题（下方）中。 @some-time就像事件时间戳。

requests并且cluster主题是基于划分的cluster_id，因此对同一个集群的请求将按顺序排列，并且可以按顺序处理。我正在使用 Kafka Streams。

如果请求之间的间隔至少为 50-100 毫秒，我很好。

但。假设有一些连续的（在几毫秒内）创建 VM 的请求

如果我使用requestsKStream 中的事件并将它们与clusterKTable 连接，并在调度 VM 后发布新的集群状态cluster，那么第二个请求将不会看到此更新，因为它比集群更新事件（并读取第二个请求比推送集群更新然后使用它更快）。

每个请求都会看到前一个请求的集群更新。无论是通过延迟请求处理还是任何其他方式，这都是我想要的。

我希望卡夫卡已经有一个机制来做类似的事情，你可以给我指点！

这是我的猜测：

requests向和cluster主题添加元数据。即事件cluster中将包含last_request_id- 最后处理的请求。 last_request_id也将存储在线程局部变量中并传递给下一个请求。请求被丰富last_request_id并转发到新delayed-requests主题。然后有可能加入cluster并delayed-requests继续last_request_id处理。
使用有关给定分区中集群的数据创建实例前瞬态状态存储（in-mem？）。请求读取和写入此存储，以及发布到cluster主题 - 持久存储。在启动状态存储从cluster主题启动。

UPD

看看这个问题，会尝试，希望这对我有用

score 0 · Accepted Answer

使用 DSL，可能无法实现您想要的。我想您可以使用处理器 API：对于这种情况，您使用两个状态存储，一个用于“使用表”，第二个用于缓冲“供应请求”。

每次处理“供应请求”时，您都会将“使用表”标记为被阻止（可能使用一些特殊键）。如果第二个“供应请求”进入并且表被“阻塞”，则将其缓冲在“请求缓冲存储”中。每次更新“使用表”时，您都会检查“请求缓冲存储”中是否有任何缓冲事件并处理一个请求。如果请求存储为空，您可以“解锁”该表。