Hi James et all, Thanks for your help. It works well when that parameter, but only for one CSV file. If I run >=5 CSV files, each of size 110MB, the data is lost too (when I check the number of received messages and the number of messages of original files) I get many lots of errors after re-running producers:
[2017-01-05 10:18:55,110] ERROR Error when sending message to topic MyTopic with key: null, value: 380 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException: Batch containing 36 record(s) expired due to timeout while requesting metadata from brokers for MyTopic-0 [2017-01-05 10:18:55,110] ERROR Error when sending message to topic MyTopic with key: null, value: 373 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.TimeoutException: Batch containing 36 record(s) expired due to timeout while requesting metadata from brokers for MyTopic-0 Could you please help me fix this problem? Thanks a lot. Best regards, Alex On Thu, Jan 5, 2017 at 9:53 AM, James Cheng <[email protected]> wrote: > kafka-console-producer.sh defaults to acks=0, which means that the > producer essentially throws messages at the broker and doesn't wait/retry > to make sure they are properly received. > > In the kafka-console-producer.sh usage text: > --request-required-acks <Integer: The required acks of the producer > request required acks> requests (default: 0) > > Try re-running your test with "--request-required-acks -1" or > "--request-required-acks all" (They are equivalent) This will tell the > broker to wait for messages to be fully saved to all replicas before > returning an acknowledgement to the producer. You can read more about acks > in the producer configuration section of the kafka docs ( > http://kafka.apache.org/documentation/#producerconfigs < > http://kafka.apache.org/documentation/#producerconfigs>) > > -James > > > On Jan 4, 2017, at 1:25 AM, Hoang Bao Thien <[email protected]> > wrote: > > > > Hi all, > > > > I have a problem with losing messages from Kafka. > > The situation is as follows: I put a csv file with 286701 rows (size = > > 110MB) into Kafka producer with command: > > $ cat test.csv | kafka-console-producer.sh --broker-list localhost:9092 > > --topic MyTopic > /dev/null > > > > and then count the number of lines from the Kafka consumer > > (kafka-console-consumer.sh --zookeeper localhost:2181 --topic MyTopic > > --from-beginning) > > However, I only get about 260K-270K, and this number of received messages > > changes for each test. > > > > My configuration in the "config/server.properties" has some minor change > > compared to the original file: > > > > log.retention.check.interval.hours=24 > > log.retention.hours=168 > > delete.topic.enable = true > > > > The remaining configurations are the same as default value. > > > > Could you please explain why the messages were lost in Kafka? And how to > > fix this problem please? > > > > Thanks a lot. > > > > Best regards > > , > > Alex > >
