4 ms·

It is a little surprising, and I agree, the docs here are not doing a particularly good job of explaining it. It might help to ask: if you don't explicitly comm

• dangoodmanUT 2y ago
  This, it has no idea that you processed the message. It assumes processing is successful by default which is cosmically stupid.
• williamdclt 2y ago
  > if you don't explicitly commit, how does Kafka know when you've processed the messages it gave you? I did expect that auto-commit still involved an explicit commit. I expected that it meant that the consumer side would commit _after_ processing a message/batch _if_ it had been >= autocommit_interval since the last commit. In other words, that it was a functionality baked into the Kafka client library (which does know when a message has been processed by the application). I don't know if it really makes sense, I never really thought hard about it before! I'm still a bit skeptical... I'm pretty sure (although not positive) that I've seen consumers with autocommit being stuck because of timeouts that were much greater than the autocommit interval, and yet retrying the same message in a loop
  • deleted 2y ago
    [deleted]
  • aphyr 2y ago
    Here's a good article from New Relic on the problem, if you'd like more detail: https://newrelic.com/blog/best-practices/kafka-consumer-config-auto-commit-data-loss#kafka-consumer-auto-commit-mitigating-data-loss-and-duplication https://newrelic.com/blog/best-practices/kafka-consumer-conf... Or here, you can reproduce it yourself using the Bufstream or Redpanda/Kafka test suite. Here's a real quick run I just dashed off. You can watch it skip over writes: https://gist.github.com/aphyr/1af2c4eef9aacde7f08f158230490855 https://gist.github.com/aphyr/1af2c4eef9aacde7f08f1582304908... lein run test --enable-auto-commit --bin bufstream-0.1.3-rc.12 --time-limit 30 --txn --final-time-limit 1/10000
• justinsaccount 2y ago
  Weird, I would have guessed that it auto commits the previous batch when it polls for the next batch, meaning it would be like loop: messages = poll() # poll returns new messages and commits previous batch process(messages) but it sounds like it "poll returns new messages and immediately commits them."
  • williamdclt 2y ago
    Information on the internet about this seems unreliable, confusing and contradictory... It's crazy for something so critical, especially when it's enabled by default.
• jakewins 2y ago
  Auto commit has always seemed super shady. Manual commit I have assumed is safe though - something something vector clocks - and it’d be really interesting to know if that trust is misplaced. What is the process and cost for having you do a Jepsen test for something like that?
  • aphyr 2y ago
    You'll find lots about the Jepsen analysis process here: https://jepsen.io/services/analysis https://jepsen.io/services/analysis
• frant-hartm 2y ago
  > how does Kafka know when you've processed the messages it gave you? By calling `poll()` again. It doesn't commit the records returned from poll until auto commit interval expires AND you call poll again. At least this is what the javadoc says quite clearly: https://kafka.apache.org/39/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html https://kafka.apache.org/39/javadoc/org/apache/kafka/clients... Note: Using automatic offset commits can also give you "at-least-once" delivery, but the requirement is that you must consume all data returned from each call to poll(Duration) before any subsequent calls, or before closing the consumer. E.g. the following commits every 10s - on each call to `poll`, it doesn't automagically commit every 5 s. Properties props = new Properties(); props.setProperty("bootstrap.servers", "localhost:9092"); props.setProperty("group.id", "test"); props.setProperty("enable.auto.commit", "true"); props.setProperty("auto.commit.interval.ms", "5000"); props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("my-topic")); while (true) { ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100)); for (ConsumerRecord<String, String> record : records) { System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value()); } Thread.sleep(10_000); }
  • frant-hartm 2y ago
    Just a note: I am not claiming it is working correctly, only saying there is a clear and documented way how the client knows when to commit, and that it works as expected in a simple scenario.
• aphyr 2y ago
  I (and apparently the Confluent docs?) may be wrong about this. I've added an update to the report.