Сериализатор несовместим с фактическим типом ключа (тип ключа: java .lang.String) - PullRequest
Новая
       33

Сериализатор несовместим с фактическим типом ключа (тип ключа: java .lang.String)

0 голосов
/ 02 мая 2020

Я пытался присоединиться к потоку с GlobaKTable, получая проблему с сериализацией. Объединение productStream KStream<String,String> с productTable равно GlobalKTable<String,Product> (здесь Product является сообщением AVRO). Ниже приведен фрагмент кода, 

KStream<String, Product> detailedProductStream = productStream.join(
                productTable,
                (productStreamKey, productStreamValue) -> productStreamValue,
                (productStreamValue, productTableValue) ->  productTableValue
);

Определено SpecificAvroSerde для сообщения Avro и Serdes.String () для Строковое сообщение. Я получаю ниже исключения. Любая подсказка? Спасибо 

Exception in thread "web-click-charts-a086d30c-5f8d-4e1f-8e3e-ab4b2ac8f03a-StreamThread-1" org.apache.kafka.streams.errors.StreamsException: Exception caught in process. taskId=0_0, processor=KSTREAM-SOURCE-0000000006, topic=PRODUCT-CURRENT, partition=0, offset=2, stacktrace=org.apache.kafka.streams.errors.StreamsException: A serializer (org.apache.kafka.common.serialization.ByteArraySerializer) is not compatible to the actual key type (key type: java.lang.String). Change the default Serdes in StreamConfig or provide correct Serdes via method parameters.
            at org.apache.kafka.streams.state.StateSerdes.rawKey(StateSerdes.java:175)
            at org.apache.kafka.streams.state.internals.MeteredWindowStore.keyBytes(MeteredWindowStore.java:222)
            at org.apache.kafka.streams.state.internals.MeteredWindowStore.fetch(MeteredWindowStore.java:167)
            at org.apache.kafka.streams.processor.internals.ProcessorContextImpl$WindowStoreReadWriteDecorator.fetch(ProcessorContextImpl.java:547)
            at org.apache.kafka.streams.kstream.internals.KStreamKStreamJoin$KStreamKStreamJoinProcessor.process(KStreamKStreamJoin.java:94)
            at org.apache.kafka.streams.processor.internals.ProcessorNode.process(ProcessorNode.java:117)
            at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:201)
            at org.apache.kafka.streams.processor.internals.ProcessorContextImpl.forward(ProcessorContextImpl.java:180)

Serdes определен как показано ниже 

Map<String, String> props = new HashMap<>();
props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

SpecificAvroSerde<Product> productSerde = new SpecificAvroSerde<>();
productSerde.configure(props, false);

Serde stringSerde = Serdes.String();

GlobalKTable<String, Product> productTable = builder.globalTable("TOPIC-1",
                Consumed.with(stringSerde, productSerde));
KStream<String, String> productStream = builder.stream("TOPIC-2",
        Consumed.with(stringSerde, stringSerde));

1 Ответ

1 голос
/ 03 мая 2020

Как / где вы указали serdes? Ваш фрагмент кода не показывает эту часть.

Однако трассировка стека указывает на то, что проблема не в соединении stream-globalTable, а в оконном соединении поток-поток (трассировка стека показывает KStreamKStreamJoin) и сообщение об ошибке показывает, что ByteArraySerializer используется для строкового ключа.

...