Spring Cloud Stream partitionKeyExpression计算错误

Question

我有一个使用 Kafka 的基于 Spring Cloud Stream 的微服务。

我创建了一个有 4 个分区的 kafka 主题。

我在我的 yml 中配置了以下内容：

spring:
  cloud:
    stream:
      bindings:
        SYNC_TABLE:
          content-type: application/json
          partitionKeyExpression: payload.partitionKey
          partitionCount: 4
          destination: ${envTopicPrefix}.LEGACY_TABLE

在我的代码中，我的消息类包含（在其超类中）partitionKey 变量：

@Data
@EqualsAndHashCode(callSuper=true)
@ToString(callSuper=true)
public class TransactionResponse extends GeneralOutputMessage{

}

---

@Data
@ToString
public class GeneralOutputMessage {

    private String operationType;
    private List<String> affectedFields;
    private Object data;
    private String eventId;
    private String eventName;
    private String partitionKey; 
}

我将 TransactionsResponse 对象作为消息发送：

final TransactionResponse transactionResponse = handler.handleEvent(event);
if (transactionResponse != null) {
    outputChannels.tableSync().send(MessageBuilder.withPayload(transactionResponse).build());
    log.info("Message Sent: {}", transactionResponse);
}

我的期望是spring cloud stream会获取keypayload.partitionKey，计算它的hashCode() % 4，并将事件发送到那个partition。

但是，逻辑是完全随机的。以下是几个例子：

Math.abs("111615631".hashCode()%4) = 1。但是，消息被发送到分区号 3。

Math.abs("110019882".hashCode()%4) = 2。但是，消息被发送到分区号 0。

Math.abs("943152574".hashCode()%4) = 0。这条消息确实发送到分区号 0。

Math.abs("943198862".hashCode()%4) = 0。但是，此消息发送到分区号 2。

我正在使用 Dalston.SR1 发布系列。

我在这里错过了什么？

谢谢。

更新：

刚刚尝试使用相同的 partitionKey 发送相同的事件（但消息正文略有不同）。即使分区键相同，消息也会发送到两个不同的分区。看起来 Spring Cloud Stream 完全忽略了 partitionKeyExpression。

Answer 1

这是我的错误，我忘记在 yml 中添加 producer: 部分：

spring:
  cloud:
    stream:
      bindings:
        SYNC_TABLE:
          content-type: application/json
          producer:
            partitionKeyExpression: payload.partitionKey
            partitionCount: 4
          destination: ${envTopicPrefix}.LEGACY_TABLE