ClouderaRegistryKafkaSerializationSchema

You can construct the schema serialization with the ClouderaRegistryKafkaSerializationSchema.builder(..) object for FlinkKafkaProducer. You must set the topic configuration and RegistryAddress parameter in the object.

The serialization schema can be constructed using the ClouderaRegistryKafkaSerializationSchema.builder(..) object.

Required settings:
  • Topic configuration when creating the builder, which can be static or dynamic (extracted from the data)
  • RegistryAddress parameter on the builder to establish the connection
Optional settings:
  • Arbitrary SchemaRegistry client configuration using the setConfig method
  • Key configuration for the produced Kafka messages
    • Specifying a KeySelector function that extracts the key from each record
    • Using a Tuple2 stream for (key, value) pairs directly
  • Security configuration
KafkaSerializationSchema<ItemTransaction> schema = ClouderaRegistryKafkaSerializationSchema
  .<ItemTransaction>builder(topic)
  .setRegistryAddress(registryAddress)
  .setKey(ItemTransaction::getItemId)
  .build();
KafkaSink<ItemTransaction> kafkaSink = new FlinkKafkaProducer<>("dummy", schema, kafkaProps, FlinkKafkaProducer.Semantic.AT_LEAST_ONCE);