Kafka Stream Chained LeftJoin - 在新消息之后再次处理之前的旧消息

Question

我有一个由其他流组成的流

final KTable<Long, CompositeInfo> compositeInfoTable = compositeImcTable
    .leftJoin(
        compositeFundTable, 
        (CompositeImc cimc, CompositeFund cf) -> {
            CompositeInfo newCandidate = new CompositeInfo();
            if (cimc != null) {
                newCandidate.imcName = cimc.imcName;
                newCandidate.imcID = cimc.imcID;                                                                    
                if (cf != null) {
                    newCandidate.investments = cf.investments;
                }
            }
            return newCandidate;
        })
    .leftJoin(
        compositeGeographyTable, 
        (CompositeInfo cinfo, CompositeGeography cg) -> {
            if (cg != null) {
                cinfo.regions = cg.regions;
            }
            return cinfo;
        })
    .leftJoin(
        compositeSectorTable, 
        (CompositeInfo cinfo, CompositeSector cs) -> {
            if (cs != null) {
                cinfo.sectors = cs.sectors;
            }
            return cinfo;
        })
    .leftJoin(
        compositeClusterTable, 
        (CompositeInfo cinfo, CustomCluster cc) -> {
            if (cc != null && cc.clusters != null) {
                cinfo.clusters = cc.clusters;
            }
            return cinfo;
        })
    .leftJoin(
        compositeAlphaClusterTable, 
        (CompositeInfo cinfo, CompositeAlphaCluster cac) -> {
            if (cac != null) {
                cinfo.alphaClusters = cac.alphaClusters;
            };
            return cinfo;
        },
        Materialized.<Long, CompositeInfo, KeyValueStore<Bytes, byte[]>>as(this.storeName)
            .withKeySerde(Serdes.Long())
            .withValueSerde(compositeInfoSerde));

我的问题与 CompositeInfo 和 CustomCluster 之间的左连接有关。 CustomCluster 如下所示

KTable<Long, CustomCluster> compositeClusterTable = builder
    .stream(
        SUB_TOPIC_COMPOSITE_CLUSTER,
        Consumed.with(Serdes.Long(), compositeClusterSerde))
    .filter((k, v) -> v.clusters != null)
    .groupByKey(Serialized.with(Serdes.Long(), compositeClusterSerde))
    .reduce((aggValue, newValue) -> newValue);

自定义集群中的消息看起来像

CustomCluster [clusterId=null, clusterName=null, compositeId=280, operation=null, clusters=[Cluster [clusterId=6041, clusterName=MyName]]]

所以我将这个对象中的 HashMap 集群分配给 CompositeInfo 对象中加入到compositeId 上的集群。

我所看到的是，一个给定的compositeId 的CustomCluster 消息进来了一个dis 正确处理，但是包含前一个集群的旧消息（我仍在调查这个）被再次处理。 在挖掘问题发生在kafka内部KTableKTableRightJoin

public void process(final K key, final Change<V1> change) {
    // we do join iff keys are equal, thus, if key is null we cannot join and just ignore the record
    if (key == null) {
        return;
    }

    final R newValue;
    R oldValue = null;

    final V2 value2 = valueGetter.get(key);
    if (value2 == null) {
        return;
    }

    newValue = joiner.apply(change.newValue, value2);

    if (sendOldValues) {
        oldValue = joiner.apply(change.oldValue, value2);
    }

    context().forward(key, new Change<>(newValue, oldValue));
}

当joine第一次返回时，newValue正确更新。但是代码然后转到 sendOldValues 块，一旦加入者返回，newValue 就是更新增益，但这次是旧集群值。

所以这是我的问题：

1. 为什么当 joiner 被调用时 newValues 会被更新 第二次使用 oldValue
2. 有没有办法关闭 sendOldValues
3. 我的链式左连接是否与它有关。我知道 以前版本的 kafka 存在链接错误。但现在我在 1.0

更新： 我发现的另一件事。如果我将连接移动到连接链上并删除其他连接，则 sendOldValues 仍然为 False。因此，如果我有类似以下内容：

final KTable<Long, CompositeInfo> compositeInfoTable = compositeImcTable
    .leftJoin(
        compositeFundTable, 
        (CompositeImc cimc, CompositeFund cf) -> {
            CompositeInfo newCandidate = new CompositeInfo();
            if (cimc != null) {
                newCandidate.imcName = cimc.imcName;
                newCandidate.imcID = cimc.imcID;
                if (cf != null) {
                    newCandidate.investments = cf.investments;
                }
            }   
            return newCandidate;
        })
    .leftJoin(
        compositeClusterTable, 
        (CompositeInfo cinfo, CustomCluster cc) -> {
            if (cc != null && cc.clusters != null) {
                cinfo.clusters = cc.clusters;
            }
            return cinfo;
        },
        Materialized.<Long, CompositeInfo, KeyValueStore<Bytes, byte[]>>as(this.storeName)
          .withKeySerde(Serdes.Long())
          .withValueSerde(compositeInfoSerde));

这给了我正确的结果。但我认为，如果我在此之后放置更多的链式连接，它们可能会显示相同的错误行为。

我目前还不确定，但我认为我的问题在于 chained leftjoin 和计算 oldValue 的行为。有没有其他人遇到过这个问题？

更新

经过大量挖掘，我意识到 sendOldValues 是 kafka 内部的，而不是我遇到的问题的原因。我的问题是，当 oldValue 的 ValueJoiner 返回时 newValue 会发生变化，我不知道它是否是由于对 Java 对象的引用分配传递所致

这是传入对象的样子

CustomCluster [clusterId=null, clusterName=null, compositeId=280, operation=null, clusters=[Cluster [clusterId=6041, clusterName=Sunil 2]]]

集群是HashSet<Cluster> clusters = new HashSet<Cluster>();

然后它被连接到一个对象

CompositeInfo [compositeName=BUCKET_NM-280, compositeID=280, imcID=19651, regions=null, sectors=null, clusters=[]]

这里的簇是同类型的，但是在 CompositeInfo 类中

当我加入时，我将 CustomCluster 对象的集群分配给 CompositeInfo 对象

(CompositeInfo cinfo, CustomCluster cc) -> {
    if (cc != null && cc.clusters != null) {
        cinfo.clusters = cc.clusters;
    }
    return cinfo;
}

Answer 1

在自己偶然发现同样的问题之后，我想提供一个详细的答案以及一个有助于说明问题的简化示例。

  @Bean
  public Function<KTable<String, String>,
    Function<KTable<String, String>, Consumer<KTable<String, String>>>> processEvents() {
    return firstnames ->
      lastnames ->
        titles -> firstnames
          .mapValues(firstname -> new Salutation().withFirstname(firstname))
          .join(lastnames, (salutation, lastname) -> salutation.withLastname(lastname))
          .leftJoin(titles, (salutation, title) -> salutation.withTitle(title))
          .toStream()
          .foreach((key, salutation) -> log.info("{}: {}", key, salutation));
  }

该示例（使用 Spring Cloud Stream 和 Kafka Streams binder）显示了一种常见模式，其中主题内容被合并到一个累加器对象中。在我们的例子中，通过加入代表名字、姓氏和（可选）标题的主题，将称呼（例如“亲爱的史密斯女士”）累积/聚合到 Salutation 对象中。

需要注意的是，在这个例子中，Salutation 实例是一个可变对象，它是一步一步构造的。在运行这样一段代码的时候，你会看到在更改一个人的姓氏时，合并总是会“跑在后面”。这意味着如果您因为 Smith 女士刚刚结婚并且现在被称为“Johnson”而发布了 lastname 事件，那么 Kafka Streams 将再次发出一个代表“Ms. Smith”的Salutation，尽管事实上她改变了她的最后一个姓名。只有当您针对姓氏主题（例如“Miller”）为同一个人发布另一个事件时，才会记录“Dear Ms. Johnson”。

这种行为的原因在位于KTableKTableInnerJoin.java的一段代码中找到：

if (change.newValue != null) {
    newValue = joiner.apply(change.newValue, valueRight);
}

if (sendOldValues && change.oldValue != null) {
    oldValue = joiner.apply(change.oldValue, valueRight);
}

context().forward(key, new Change<>(newValue, oldValue), To.all().withTimestamp(resultTimestamp));

joiner 是ValueJoiner，在我们的例子中可以是是(salutation, lastname) -> salutation.withLastname(lastname)，如上图所示。这段代码的问题在于，如果您使用带有可变累加器对象（在我们的例子中是 Salutation 的实例）的累加模式，它（按设计）被重用于所有连接，那么 oldValue 和 @ 987654332@ 将是同一个对象。而且，由于oldValue是之后计算出来的，它会包含旧的姓氏，这就解释了为什么Spring Kafka落后了。

因此，ValueJoiner 返回的对象每次都是一个不包含对其他可变对象的引用的新对象至关重要，这些对象可能是共享的（因此是变异的）。因此，最安全的方法是让ValueJoiner 返回一个不可变对象。

我不会认为这是库的错误，因为它必须以某种方式比较旧状态和新状态，并且因为获取可变对象的快照需要深拷贝。但是，在文档中提及它可能是值得的。此外，在oldValue == newValue 时发出警告至少会让人们意识到这个问题。我会检查是否可以纳入这些改进。

Answer 2

这确实是一个通过引用传递的问题。加入时，我需要初始化并返回一个新对象，而不是给旧对象赋值。

_{根据fizi cmets 的问题回答。}