Htykuut

I have written a code to push data to kafka topic on daily basis, but there are few issue which i am not sure this code will be able to handle. my responsibility is to push complete data from a live table which holds 1 day data(refreshed every day morning)

my code will query "select * from mytable" and push it one by one to kafka topic as before pushing i need to validate/alter each row and push to topic.

below is my producer send code.

    Properties configProperties = new Properties();

        configProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, sBOOTSTRAP_SERVERS_CONFIG);

        configProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,

                "org.apache.kafka.common.serialization.StringSerializer");

        configProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,

                "org.apache.kafka.common.serialization.StringSerializer");

        configProperties.put("acks", "all");

        configProperties.put("retries", 0);

        configProperties.put("batch.size", 15000);

        configProperties.put("linger.ms", 1);

        configProperties.put("buffer.memory", 30000000);

        @SuppressWarnings("resource")

        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(configProperties);

        System.out.println("Starting Kafka producer job  " + new Date());

        producer.send(new ProducerRecord<String, String>(eventName, jsonRec.toString()), new Callback() {

            public void onCompletion(RecordMetadata metadata, Exception e) {

                if (e != null) {

                    e.printStackTrace();

                }

            }

        });

Now, i am not sure how to push data back again into topic in case of failure. Since i have selected all the records from table and few of it got failed and i do not know which all.

Below is what i want to address

1. how can process only those records which are not pushed to avoid duplicate record being push(avoid redundancy).




2. how to validate the records pushed are exactly same as in table. i mean the data integrity. like size of data and count of records been pushed.

You can use configProperties.put("enable.idempotence", true); - it will try to retry failed messages but make sure there will be just one of each record saved in kafka. Note that it implies that retries>0 acks=all and max.in.flight.requests.per.connection >=0. For details check https://kafka.apache.org/documentation/.

For 2nd question - if you mean that you need to save all records or none then you have to use kafka transactions, which brings a lot more questions, I would recommend reading https://www.confluent.io/blog/transactions-apache-kafka/