Thanks for sharing the full solution, much appreciated! On Thu, 10 Feb 2022 at 09:07, HG <hanspeter.sl...@gmail.com> wrote:
> The complete solution for the record ( that others can benefit from it). > > KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder() > .setProperties(kafkaProps) > .setProperty("ssl.truststore.type",trustStoreType) > .setProperty("ssl.truststore.password",trustStorePassword) > .setProperty("ssl.truststore.location",trustStoreLocation) > .setProperty("security.protocol",securityProtocol) > .setProperty("partition.discovery.interval.ms", > partitionDiscoveryIntervalMs) > .setProperty("commit.offsets.on.checkpoint", > commitOffsetsOnCheckpoint) > .setGroupId(groupId) > .setTopics(kafkaInputTopic) > .setDeserializer(KafkaRecordDeserializationSchema.of(new > JSONKeyValueDeserializationSchema(fetchMetadata))) > > .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) > .build(); > > > Op wo 9 feb. 2022 om 09:46 schreef HG <hanspeter.sl...@gmail.com>: > >> Sorry to have bothered everyone. >> >> This is the obvious solution: >> >> .setDeserializer(KafkaRecordDeserializationSchema.of(new >> JSONKeyValueDeserializationSchema(false))) >> >> >> Regards Hans-Peter >> >> >> Op di 8 feb. 2022 om 21:56 schreef Roman Khachatryan <ro...@apache.org>: >> >>> Hi, >>> >>> setDeserializer() expects KafkaRecordDeserializationSchema; >>> JSONKeyValueDeserializationSchema you provided is not compatible with >>> it. >>> You can convert it using [1] >>> >>> [1] >>> >>> https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.html#of-org.apache.flink.streaming.connectors.kafka.KafkaDeserializationSchema- >>> >>> >>> Regards, >>> Roman >>> >>> On Tue, Feb 8, 2022 at 5:43 PM HG <hanspeter.sl...@gmail.com> wrote: >>> > >>> > Hi all, >>> > >>> > When I build this code: >>> > >>> > KafkaSource<ObjectNode> source = KafkaSource.<ObjectNode>builder() >>> > .setProperties(kafkaProps) >>> > .setProperty("ssl.truststore.type",trustStoreType) >>> > .setProperty("ssl.truststore.password",trustStorePassword) >>> > .setProperty("ssl.truststore.location",trustStoreLocation) >>> > .setProperty("security.protocol",securityProtocol) >>> > .setProperty("partition.discovery.interval.ms", >>> partitionDiscoveryIntervalMs) >>> > .setProperty("commit.offsets.on.checkpoint", >>> commitOffsetsOnCheckpoint) >>> > .setGroupId(groupId) >>> > .setTopics(kafkaInputTopic) >>> > .setDeserializer(new JSONKeyValueDeserializationSchema(false)) >>> > >>> >>> .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST)) >>> > .build(); >>> > >>> > >>> > I get: >>> > This error: >>> > >>> > error: incompatible types: >>> org.apache.flink.streaming.util.serialization.JSONKeyValueDeserializationSchema >>> cannot be converted to >>> org.apache.flink.connector.kafka.source.reader.deserializer.KafkaRecordDeserializationSchema<org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode> >>> > .setDeserializer(new >>> JSONKeyValueDeserializationSchema(false)) >>> > >>> > >>> > What am I doing wrong? >>> > As per the documentation JSONKeyValueDeserializationSchema returns an >>> ObjectNode. >>> > >>> > Regards Hans >>> > >>> >>