我是Kafka的新人。我的要求是，在数据库源和目标中有两个表。现在我想从源表中获取数据并将其存储到目标表中，Kafka将作为生产者和消费者工作。我已经完成了代码，但问题是当producer生成数据时，会丢失一些要生成的数据。例如，如果我在源表中有100条记录，那么它不会生成所有100条记录。我用的是Kafka-0.10
myproducer配置-

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

我的制作人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);
    }
}

我的消费者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
消费者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");
            }
        }
    }
}