我用Kafka来处理信息。消息可以分为几个部分（它是一个复合消息）。所以在流中，我可以有一个复合消息，它被分成三部分。换句话说，Kafka流中有三条记录，但这是一条重要的信息。我想使用kafka表合并一个kafka记录中的复合消息部分。合并后，一条消息将插入数据库（postgres）。每个零件都有零件号和零件总数。例如，如果我在流中有一条消息的三个部分（三个Kafka记录），那么每个部分都有值为3的部分总数字段。
我的理解是，在积极的场景中，任务很简单：在表中聚合部分，从表中创建流，并过滤具有相同聚合部分大小和部分总数的记录，在一个合并消息中Map过滤并将其插入数据库（postgres）。
但消极的情况也是可能的。在极少数情况下，其中一个部分根本不能插入Kafka（或者它将在很长时间后，在超时后插入）。例如，在流中，一条复合消息的三个部分中只有两个部分会出现。在这种情况下，我必须在数据库（postgres）中插入未完全构造的消息（它将只包含两部分，而不是三部分）。我怎样才能在Kafka身上实现这种消极情景？