我们正在升级到Spring-Cloud-Stream的3.0.0-Release，并遇到以下问题：
使用这种功能样式时：
public class EventProcessor {
private final PriceValidator priceValidator;
@Bean
public Function<Flux<EnrichedValidationRequest>, Flux<ValidationResult>> validate() {
return enrichedValidationRequestFlux -> enrichedValidationRequestFlux
.map(ProcessingContext::new)
.flatMap(priceValidator::validateAndMap);
}
}
application.yaml看起来像这样：
spring.cloud.stream:
default-binder: kafka
kafka:
binder:
brokers: ${kafka.broker.prod}
auto-create-topics: false
function.definition: validate
# INPUT: enrichedValidationRequests
spring.cloud.stream.bindings.validate-in-0:
destination: ${kafka.topic.${spring.application.name}.input.enrichedValidationRequests}
group: ${spring.application.name}.${STAGE:NOT_SET}
consumer:
useNativeDecoding: true
spring.cloud.stream.kafka.bindings.validate-in-0:
consumer:
configuration:
key.deserializer: org.apache.kafka.common.serialization.StringDeserializer
value.deserializer: de.pricevalidator.deserializer.EnrichedValidationRequestDeserializer
# OUTPUT: validationResults
spring.cloud.stream.bindings.validate-out-0:
destination: validationResultsTmp
producer:
useNativeEncoding: true
spring.cloud.stream.kafka.bindings.validate-out-0:
producer:
compression.type: lz4
messageKeyExpression: payload.offerKey
configuration:
key.serializer: org.apache.kafka.common.serialization.StringSerializer
value.serializer: de.pricevalidator.serializer.ValidationResultSerializer
似乎序列化完成了两次-当我们拦截在kafka主题中生成的消息时，使用者只会将它们显示为JSON（字符串），但是现在这是一个不可读的字节[]。此外，生产中的下游消费者无法再反序列化消息。奇怪的是，无论我们在使用者属性中使用什么内容（无论是在活页夹上还是在默认的kafka级别上），输入消息的反序列化似乎都可以正常工作我们有种感觉，这个错误“又回来了”，但是我们无法在代码中找到确切的位置：https://github.com/spring-cloud/spring-cloud-stream/issues/1536
我们的（丑陋的）解决方法：
@Slf4j
@Configuration
public class KafkaMessageConverterConfiguration {
@ConditionalOnProperty(value = "spring.cloud.stream.default-binder", havingValue = "kafka")
@Bean
public MessageConverter validationResultConverter(BinderTypeRegistry binder, ObjectMapper objectMapper) {
return new AbstractMessageConverter(MimeType.valueOf("application/json")) {
@Override
protected boolean supports(final Class<?> clazz) {
return ValidationResult.class.isAssignableFrom(clazz);
}
@Override
protected Object convertToInternal(final Object payload, final MessageHeaders headers, final Object conversionHint) {
return payload;
}
};
}
}
是否存在一种设置自定义序列化程序或获取本机编码的“正确”方法？