How to prevent unwanted attributes in Kafka message getting inserted using Kafka connect sink?

Refresh

April 2019

Views

490 time

1

I am publishing messages into my Kafka topic using Java producer application, using the code below.

String mySchema = "{"type": "record","name": "MyData","namespace": "com.qwe.rty","doc": "MyData Schema","fields": [{"name": "f1","type": ["null", "string"],"default" : null}, {"name": "f2","type": ["null", "string"],"default" : null}, {"name": "f3","type": ["null", "string"],"default" : null}]}";


Properties props = new Properties();

props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.0.1:9092");

props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,io.confluent.kafka.serializers.KafkaAvroSerializer.class);

props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,io.confluent.kafka.serializers.KafkaAvroSerializer.class);

props.put("schema.registry.url", "http://192.168.0.1:8081");



KafkaProducer producer = new KafkaProducer(props);

Schema.Parser parser = new Schema.Parser();

Schema schema = parser.parse(mySchema);



GenericRecord avroRecord = new GenericData.Record(schema);

avroRecord.put("f1", "data1");

avroRecord.put("f2", "data2");

avroRecord.put("f3", "data3");



ProducerRecord<String, GenericRecord> record = new ProducerRecord<>("my_topic_1",avroRecord);



try {

    System.out.println("send from producer.serialized.avro.Sender6n");

    producer.send(record);

} catch (Exception e) {

    e.printStackTrace();

}



try{

    System.out.println("flush from producer.serialized.avro.Sender6n");

    producer.flush();

}catch(Exception e){

    e.printStackTrace();

}

The messages are in Avro. I am using Kafka connect jdbc sink to insert it to my Oracle table, using the configuration below;

name=jdbc-sink-avro

connector.class=io.confluent.connect.jdbc.JdbcSinkConnector

tasks.max=1

topics=my_topic_1

connection.url=jdbc:oracle:thin:@192.168.0.1:1521:user01

connection.user=user_01

connection.password=user_01

auto.create=true

table.name.format=my_table_1

key.converter=io.confluent.connect.avro.AvroConverter

key.converter.schema.registry.url=http://192.168.0.1:8081

value.converter=io.confluent.connect.avro.AvroConverter

value.converter.schema.registry.url=http://192.168.0.1:8081

producer.retries=1

This works fine. But I don't want f3 to get inserted to my table, which means, I just need a 2 column table. In other words, I want to prevent f3 getting inserted. How can I do this?

1 answers

1

Use Kafka Connect's Single Message Transform (SMT) functionality. Specifically ReplaceField and blacklist.

You can see an example here ("Field Masking and Whitelist/Blacklists").