如何利用Spring-Cloud-Stream实现自定义kafka分区

Question

我试图使用spring cloud stream绑定实现一个自定义的Kafka分区器。我只想对用户主题进行自定义分区，而不对公司主题进行任何操作（在本例中，Kafka将使用DefaultPartitioner）。

我的绑定配置：

spring:
  cloud:
    stream:
      bindings:
        comp-out:
          destination: company
          contentType: application/json
        user-out:
          destination: user
          contentType: application/json

spring:
  cloud:
    stream:
      bindings:
        comp-out:
          destination: company
          contentType: application/json
        user-out:
          destination: user
          contentType: application/json
		      producer:
            partitioned: true
            partitionSelectorClass: config.UserPartitioner

我使用以下方式将消息发送到流中：

public void postUserStream(User user) throws ServiceException {
        try {
            LOG.info("Posting User {} into Kafka stream...", user);
            MessageChannel messageChannel = messageStreams.outboundUser();
            messageChannel
                    .send(MessageBuilder.withPayload(user)
                            .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON).build());
        } catch (Exception ex) {
            LOG.error("Error while populating User stream into Kafka.. ", ex);
            throw ex;
        }
    }

我的UserPartitioner类：

public class UserPartitioner extends DefaultPartitioner {
    @Override
    public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes,
          Cluster cluster) {

        String partitionKey = null;
        if (Objects.nonNull(value)) {
            User user = (User) value;
            partitionKey = String.valueOf(user.getCompanyId()) + "_" + String.valueOf(user.getId());
            keyBytes = partitionKey.getBytes();
        }
        return super.partition(topic, partitionKey, keyBytes, value, valueBytes, cluster);
    }
}

我最终收到以下异常：

Property: spring.cloud.stream.bindings.user-out.producer.partitioned
Value: true
Origin: "spring.cloud.stream.bindings.user-out.producer.partitioned" from property source "bootstrapProperties"
Reason: No setter found for property: partitioned

编辑：根据文档，还尝试了以下步骤：

User-Out：Destination：user ContentType：Application/JSON Producer：PartitionKeyExtractorClass：

@Component
public class SimpleUserPartitioner implements PartitionKeyExtractorStrategy {

	@Override
	public Object extractKey(Message<?> message) {
		if(message.getPayload() instanceof BaseUser) {
			BaseUser user = (BaseUser) message.getPayload();
			return user.getId();
		}
		return 10;
	}
    
}

更新2：适用于我的解决方案将partitioncount添加到绑定中，并在绑定器中将autoaddpartitions添加到true中：

spring:
  logging:
    level: info
  cloud:
    stream:
      bindings:
        user-out:
          destination: user
          contentType: application/json
          producer:
            partition-key-expression: headers['partitionKey']
            partition-count: 4

spring:
  cloud:
    stream:
      kafka:
        binder:
          brokers: localhost:9092
          autoAddPartitions: true

Answer 1

没有属性分区；吸气器取决于其他属性...

public boolean isPartitioned() {
    return this.partitionKeyExpression != null
            || this.partitionKeyExtractorName != null;
}

partitionSelectorClass: config.UserPartitioner

UserPartitioner是一个KafkaPartitioner-它决定哪些使用者获得哪些分区（在使用者端）

PartitionSelectorClass必须是PartitionSelectorStrategy-它决定将记录发送到哪个分区（在生成器端）。

鉴于您的问题，我认为您只是将Partitione与PartitionSelectorStrategy混淆了，而您需要后者。