Hi all,
When I build this code:
KafkaSource source = KafkaSource.builder()
.setProperties(kafkaProps)
.setProperty("ssl.truststore.type",trustStoreType)
.setProperty("ssl.truststore.password",trustStorePassword)
.setProperty("ssl.truststore.location",trustStoreLoca
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/Kafk
Hi
Convert ??
How does that work?
Can you spare a couple of lines for that?
Regards Hans
Op di 8 feb. 2022 om 21:56 schreef Roman Khachatryan :
> Hi,
>
> setDeserializer() expects KafkaRecordDeserializationSchema;
> JSONKeyValueDeserializationSchema you provided is not compatible with
> it.
> Yo
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 :
> Hi,
>
> setDeserializer() expects KafkaRecordDeserial
The complete solution for the record ( that others can benefit from it).
KafkaSource source = KafkaSource.builder()
.setProperties(kafkaProps)
.setProperty("ssl.truststore.type",trustStoreType)
.setProperty("ssl.truststore.password",trustStorePassword)
.setProperty(
Thanks for sharing the full solution, much appreciated!
On Thu, 10 Feb 2022 at 09:07, HG wrote:
> The complete solution for the record ( that others can benefit from it).
>
> KafkaSource source = KafkaSource.builder()
> .setProperties(kafkaProps)
> .setProperty("ssl.truststore.ty
