泉Could Steam粘合剂

Question

任何想法都将受到赞赏，我正在尝试编写测试为；

@ExtendWith(SpringExtension.class)
@EmbeddedKafka(count = 1, controlledShutdown = true, topics = { "input", "output" }, brokerProperties = { "broker.id=2",
    "listeners=PLAINTEXT://127.0.0.1:9092" })
class BindingTest {

@Autowired
private ApplicationContext applicationContext;

@Autowired
private EmbeddedKafkaBroker embeddedKafka;

@Autowired
private CustomBindings cBindings;

/**
 * @throws java.lang.Exception
 */
@BeforeEach
void setUp() throws Exception {
}

/**
 * @throws java.lang.Exception
 */
@AfterEach
void tearDown() throws Exception {
    embeddedKafka.getKafkaServers().forEach(b -> b.shutdown());
}

@Test
void test0() {
    String KEY = "KEY";
    String testMessage = "TESTMESSAGE";
    Message<String> message = MessageBuilder.withPayload(testMessage)
            .setHeader(KafkaHeaders.MESSAGE_KEY, KEY).build();
    cBindings.output().send(message);

}

@SpringBootApplication
@EnableBinding(CustomBindings.class)
public static class BindingApplication {

}

}

Spring.Cloud.Stream.Kafka.Streams.Bindings.Output.Producer.KeySerde==org.Apache.Kafka.Common.Serialization.StringSerializer Spring.Cloud.Stream.Kafka.Streams.Bindings.Output.Producer.ValueSerde==org.Apache.Kafka.Common.Serialization.StringSerializer

仍然得到

消息处理程序[org.springframework.cloud.stream.binder.kafka.kafkaMessageChannelBinder$ProducerConfigurationMessageHandler@71F0806b]中出错；嵌套html" target="_blank">异常为org.apache.kafka.Common.Errors.SerializationException：无法将类[B的值转换为value.Serializer中指定的类org.apache.kafka.Common.Serialization.StringSerializer,FailedMessage=GenericMessage[Payload=Byte[8],Headers={id=C91897BC-2E4E-0A74-BC05-17FB31B690F6,kafka_MessageKey=Key,ContentType=Application/JSON,

这对我来说没有意义

Answer 1

从外观上看，这不是一个Kafka Streams应用程序，而是一个带有Kafka绑定器的常规Spring Cloud Stream应用程序。因此，您不需要这两个属性。spring.cloud.stream.kafka.streams.bindings.output.producer.keyserde==org.apache.kafka.common.serialization.StringSerializer spring.cloud.stream.kafka.streams.bindings.output.producer.valueserde==org.apache.kafka.common.serialization.StringSerializer

此外，为了修复错误，您需要从配置中删除这一行:spring.cloud.stream.bindings.output.producer.use-native-encoding=true。

通过将native encoding设置为true，您要求Kafka执行序列化，这将依赖于默认的ByteArraySerializer。如果您真正想要进行本机序列化，则需要设置适当的值序列化程序(stringserializer)。但是由于这是一个测试，我建议您移除这个属性，看看您的测试是否通过。