如果可以使用密钥集为同一主题生成所有事件，则可以使用ksqldb中的聚合来收集特定密钥的所有事件，例如：

CREATE STREAM source (
    KEY INT KEY,  -- example key to group by
    EVENT STRING  -- example event to collect
  ) WITH (
   kafka_topic='source', -- or whatever your source topic is called.
   value_format='json' -- or whatever value format you need.
);

CREATE TABLE agg AS
  SELECT 
    key, 
    COLLECT_LIST(event) as events
  FROM source
  GROUP BY key;

这将创建一个名为 AGG 默认情况下。当接收到源主题上特定键的新事件时，ksqldb将向 AGG 主题，并将键设置为 key 以及包含所有 events 找到那把钥匙。
然后可以将此变更日志作为流导入：

CREATE STREAM agg_stream (
   KEY INT KEY,
   EVENTS ARRAY<STRING> 
) WITH (
   kafka_topic='AGG',
   value_format='json'
);

然后，您可以应用一些条件来过滤流，以便只包含最终结果：

STREAM competed_results AS 
  SELECT 
    * 
  FROM agg_stream 
  WHERE ARRAY_LEN(EVENTS) = 5; -- example 'complete' criteria.

您甚至可以使用用户定义的函数来定义完整的条件：

STREAM competed_results AS 
  SELECT 
    * 
  FROM agg_stream 
  WHERE IS_COMPLETE(EVENTS);