Apache Kafka: Producer Not Producing All Data

Question

I am new in kafka. My requirement is, I have two table in database source and destination. Now I want to fetch data from source table and store it into destination between these kafka will be work as a producer and consumer. I have done the code but problem is that when producer produces the data some data are missed to produce. For example if I have 100 records in source table then it's not produces all 100 records. I am using Kafka-0.10

MyProducer Config-

bootstrap.servers=192.168.1.XXX:9092,192.168.1.XXX:9093,192.168.1.XXX:9094
acks=all
retries=2
batch.size=16384
linger.ms=2
buffer.memory=33554432
key.serializer=org.apache.kafka.common.serialization.IntegerSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer

My Producer Code:-

public void run() {
    SourceDAO sourceDAO = new SourceDAO();
    Source source;
    int id;
    try {
        logger.debug("INSIDE RUN");
        List<Source> listOfEmployee = sourceDAO.getAllSource();
        Iterator<Source> sourceIterator = listOfEmployee.iterator();
        String sourceJson;
        Gson gson = new Gson();
        while(sourceIterator.hasNext()) {
            source = sourceIterator.next();
            sourceJson = gson.toJson(source);
            id = source.getId();
            producerRecord = new ProducerRecord<Integer, String>(TOPIC, id, sourceJson);
            producerRecords.add(producerRecord);
        }

        for(ProducerRecord<Integer, String> record : producerRecords) {
            logger.debug("Producer Record: " + record.value());
            producer.send(record, new Callback() {
                @Override
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    logger.debug("Exception: " + exception);
                    if (exception != null)
                        throw new RuntimeException(exception.getMessage());
                    logger.info("The offset of the record we just sent is: " + metadata.offset()
                            + " In Partition : " + metadata.partition());
                }
            });
        }
        producer.close();
        producer.flush();
        logger.info("Size of Record: " + producerRecords.size());
    } catch (SourceServiceException e) {
        logger.error("Unable to Produce data...", e);
        throw new RuntimeException("Unable to Produce data...", e);
    }
}

My Consumer Config:-

bootstrap.servers=192.168.1.XXX:9092,192.168.1.231:XXX,192.168.1.232:XXX
group.id=consume
client.id=C1
enable.auto.commit=true
auto.commit.interval.ms=1000
max.partition.fetch.bytes=10485760
session.timeout.ms=35000
consumer.timeout.ms=35000
auto.offset.reset=earliest
message.max.bytes=10000000
key.deserializer=org.apache.kafka.common.serialization.IntegerDeserializer

value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

Consumer Code:-

public void doWork() {
    logger.debug("Inside doWork of DestinationConsumer");
    DestinationDAO destinationDAO = new DestinationDAO();
    consumer.subscribe(Collections.singletonList(this.TOPIC));
    while(true) {
        ConsumerRecords<String, String> consumerRecords = consumer.poll(1000);
        int minBatchSize = 1;
        for(ConsumerRecord<String, String> rec : consumerRecords) {
            logger.debug("Consumer Recieved Record: " + rec);
            consumerRecordsList.add(rec);
        }
        logger.debug("Record Size: " + consumerRecordsList.size());
        if(consumerRecordsList.size() >= minBatchSize) {
            try {
                destinationDAO.insertSourceDataIntoDestination(consumerRecordsList);
            } catch (DestinationServiceException e) {
                logger.error("Unable to update destination table");
            }
        }
    }
}

Answer 1

From what could be seens here I would guess that you did not flush or close the producer. You should note that send runs async and just prepare a batch which is send later on (depending on the configuration of your producer):

From the kafka documentation

The send() method is asynchronous. When called it adds the record to a buffer of pending record sends and immediately returns. This allows the producer to batch together individual records for efficiency.

What you should try is to call producer.close() after you iterated over all producerRecords (BTW: why are you caching the entire producerRecords that might causes problems when you have to many records).

If that does not help you should try to use a e.g. a console consumer to figure out what is missing. Please offer some more code. How is the producer configured? How does your consumer look like? What is the type of producerRecords?

Hope that helps.