m1a2st commented on code in PR #23713:
URL: https://github.com/apache/kafka/pull/23713#discussion_r4185936733
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -191,17 +191,19 @@ public class ConsumerConfig extends AbstractConfig {
"<p>Note that increasing a topic's partition count while this
config is set to <code>latest</code> may cause silent " +
"message loss: producers may begin appending records to a newly
created partition before the consumer discovers it, " +
"and <code>latest</code> resets the position to the log end
offset, skipping any records produced during that discovery gap.</p>" +
- "<p>To avoid this, prefer
<code>by_duration:<duration></code>. When a partition has no committed
offset, " +
+ "<p>To reduce this risk, consider
<code>by_duration:<duration></code>. When a partition has no committed
offset, " +
"<code>by_duration</code> determines the starting position by
issuing a <code>ListOffsets</code> lookup for " +
- "<code>now() - duration</code>. If the target timestamp is earlier
than the partition's creation time, the lookup " +
- "returns the partition's start offset, ensuring that records
produced during the discovery window are still consumed. Size the duration to
cover " +
- "the worst-case partition-discovery latency for the group protocol
in use:</p>" +
- "<ul><li>With the <code>consumer</code> group protocol (KIP-848),
newly assigned partitions are pushed on the next " +
- "group heartbeat, so a value at least as large as
<code>group.consumer.heartbeat.interval.ms</code> " +
- "(server default 5000 ms) is sufficient, for example
<code>by_duration:PT5S</code>.</li>" +
+ "<code>now() - duration</code>. The lookup uses record timestamps,
not the partition's creation time. " +
+ "With <code>CreateTime</code>, newly appended records can have
timestamps older than the target and may be skipped. " +
+ "Size the duration to cover record timestamp lag as well as
partition-discovery and assignment delays:</p>" +
+ "<ul><li>With the <code>consumer</code> group protocol (KIP-848),
assignments are delivered through group heartbeats. " +
+ "Account for <code>group.consumer.heartbeat.interval.ms</code>
(server default 5000 ms) " +
+ "and metadata propagation and assignment delays.</li>" +
"<li>With the <code>classic</code> group protocol, new partitions
are discovered through periodic metadata refresh " +
- "and a subsequent rebalance, so the duration must exceed
<code>metadata.max.age.ms</code> (client default " +
- "300000 ms) plus the rebalance time, for example
<code>by_duration:PT6M</code>.</li></ul>" +
+ "and a subsequent rebalance. Account for
<code>metadata.max.age.ms</code> (client default " +
+ "300000 ms) and the rebalance time.</li></ul>" +
+ "<p>A duration based only on discovery latency does not guarantee
that all records in a new partition will be consumed. " +
+ "Use <code>earliest</code> when an offset reset must start at the
log start offset, accepting that retained records may be replayed.</p>" +
Review Comment:
Why is this change needed? The previous wording seems clearer to me, and I'm
not sure this revision improves readability.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -191,17 +191,19 @@ public class ConsumerConfig extends AbstractConfig {
"<p>Note that increasing a topic's partition count while this
config is set to <code>latest</code> may cause silent " +
"message loss: producers may begin appending records to a newly
created partition before the consumer discovers it, " +
"and <code>latest</code> resets the position to the log end
offset, skipping any records produced during that discovery gap.</p>" +
- "<p>To avoid this, prefer
<code>by_duration:<duration></code>. When a partition has no committed
offset, " +
+ "<p>To reduce this risk, consider
<code>by_duration:<duration></code>. When a partition has no committed
offset, " +
"<code>by_duration</code> determines the starting position by
issuing a <code>ListOffsets</code> lookup for " +
- "<code>now() - duration</code>. If the target timestamp is earlier
than the partition's creation time, the lookup " +
- "returns the partition's start offset, ensuring that records
produced during the discovery window are still consumed. Size the duration to
cover " +
- "the worst-case partition-discovery latency for the group protocol
in use:</p>" +
- "<ul><li>With the <code>consumer</code> group protocol (KIP-848),
newly assigned partitions are pushed on the next " +
- "group heartbeat, so a value at least as large as
<code>group.consumer.heartbeat.interval.ms</code> " +
- "(server default 5000 ms) is sufficient, for example
<code>by_duration:PT5S</code>.</li>" +
+ "<code>now() - duration</code>. The lookup uses record timestamps,
not the partition's creation time. " +
+ "With <code>CreateTime</code>, newly appended records can have
timestamps older than the target and may be skipped. " +
Review Comment:
I don't think mentioning the partition's creation time adds much here. Kafka
doesn't record it anywhere, so it's not a comparison readers would naturally
make. Could we fold this into the previous sentence and state the CreateTime
consequence directly?
```suggestion
"<code>by_duration</code> determines the starting
position by issuing a <code>ListOffsets</code> lookup for " +
"<code>now() - duration</code>, which returns the first record
whose timestamp is at or after that target. " +
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
