Я пытаюсь принять jsonSerde в качестве входных данных от тем и должен обработать запись и должен поместить ее как сообщение Avro в различных темах, используя потоки kafka. Выходные данные выглядят как двоичные, а данные не в реальном формате JSON.Как будто он использует bytearrayserde по умолчанию для значения и ключа. Я не знаю почему, но я предоставляю сериализатор как SpecificAvroSerde.
private final static JsonSerde<JsonNode> jsonSerde = new JsonSerde<JsonNode>(JsonNode.class);
private static Map<String, Object> props;
//Serde of specific record
private static SpecificAvroSerde<SpecificRecord> productValueSerde;
@Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
public StreamsConfig kafkaStreamsConfig()
throws UnknownHostException {
props = new HashMap<>();
props.put(StreamsConfig.APPLICATION_ID_CONFIG,
"*****processor-3");
props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:29092,localhost:19092,localhost:39092");
props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, SpecificAvroSerde.class);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
"http://localhost:18081");
props.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, "org.apache.kafka.clients"
+ ".consumer.RoundRobinAssignor");
productValueSerde = new SpecificAvroSerde<SpecificRecord>();
productValueSerde.configure((Collections.singletonMap(
AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
"http://localhost:18081")),false);
return new StreamsConfig(props);
}
@Bean
public KStream<JsonNode,JsonNode> KStream(StreamsBuilder kStreamBuilder){
KStream<JsonNode,JsonNode> stream = kStreamBuilder.stream("localtest",Consumed.with(jsonSerde, jsonSerde));
try {
KStream<JsonNode,SpecificRecord> avroStream = stream.flatMap((K,V)->actNationalPaperHelper.mapToCoreAvro(K, V));
//avroStream.flatMap((K,V)->System.out.println(V); return avroStream));
avroStream.through("serdetest16",Produced.with(jsonSerde, productValueSerde));
}
catch(Exception e) {
System.out.println(e);
}
return stream;
}
T