You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jeffkbkim (via GitHub)" <gi...@apache.org> on 2023/11/29 16:32:48 UTC
[PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
jeffkbkim opened a new pull request, #14870:
URL: https://github.com/apache/kafka/pull/14870
The group expiration log becomes noisy when we encounter a retry-able error as the retry backoff is fixed to 500 ms. Allow customizable retry backoff so that even in the case of failure we have a longer delay.
The current default for offsetsRetentionCheckIntervalMs is set to 10 minutes so even if the operation fails we will "retry" after 10 minutes.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
Posted by "jeffkbkim (via GitHub)" <gi...@apache.org>.
jeffkbkim commented on code in PR #14870:
URL: https://github.com/apache/kafka/pull/14870#discussion_r1412583010
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorTimer.java:
##########
@@ -52,6 +52,20 @@ interface TimeoutOperation<T, U> {
*/
void schedule(String key, long delay, TimeUnit unit, boolean retry, TimeoutOperation<T, U> operation);
+ /**
+ * Add an operation to the timer. If an operation with the same key
+ * already exists, replace it with the new operation.
+ *
+ * @param key The key to identify this operation.
+ * @param delay The delay to wait before expiring.
+ * @param unit The delay unit.
+ * @param retry A boolean indicating whether the operation should
+ * be retried on failure.
+ * @param retryBackoff The delay when rescheduled on retry.
+ * @param operation The operation to perform upon expiration.
+ */
+ void schedule(String key, long delay, TimeUnit unit, boolean retry, long retryBackoff, TimeoutOperation<T, U> operation);
Review Comment:
yes. i updated the javadocs
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14870:
URL: https://github.com/apache/kafka/pull/14870#discussion_r1413587663
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -518,9 +526,13 @@ public CoordinatorResult<OffsetDeleteResponseData, Record> deleteOffsets(
public CoordinatorResult<Void, Record> cleanupGroupMetadata() {
List<Record> records = new ArrayList<>();
groupMetadataManager.groupIds().forEach(groupId -> {
- if (offsetMetadataManager.cleanupExpiredOffsets(groupId, records)) {
+ long startMs = time.milliseconds();
+ boolean allOffsetsExpired = offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
+ if (allOffsetsExpired) {
groupMetadataManager.maybeDeleteGroup(groupId, records);
}
+ log.info("[GroupId {}] Generated {} tombstone records (allOffsetsExpired={}) in {} milliseconds.",
+ groupId, records.size(), allOffsetsExpired, time.milliseconds() - startMs);
Review Comment:
Hum... I thought that we would log only once for the entire `cleanupGroupMetadata`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -610,6 +610,7 @@ public boolean cleanupExpiredOffsets(String groupId, List<Record> records) {
partitions.forEach((partition, offsetAndMetadata) -> {
if (condition.isOffsetExpired(offsetAndMetadata, currentTimestampMs, config.offsetsRetentionMs)) {
expiredPartitions.add(appendOffsetCommitTombstone(groupId, topic, partition, records).toString());
+ log.debug("[GroupId {}] Expired offset for tp={}-{}", groupId, topic, partition);
Review Comment:
nit: `for partition`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14870:
URL: https://github.com/apache/kafka/pull/14870#discussion_r1414399614
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -524,17 +524,16 @@ public CoordinatorResult<OffsetDeleteResponseData, Record> deleteOffsets(
* @return The list of tombstones (offset commit and group metadata) to append.
*/
public CoordinatorResult<Void, Record> cleanupGroupMetadata() {
+ long startMs = time.milliseconds();
List<Record> records = new ArrayList<>();
groupMetadataManager.groupIds().forEach(groupId -> {
- long startMs = time.milliseconds();
boolean allOffsetsExpired = offsetMetadataManager.cleanupExpiredOffsets(groupId, records);
if (allOffsetsExpired) {
groupMetadataManager.maybeDeleteGroup(groupId, records);
}
- log.info("[GroupId {}] Generated {} tombstone records (allOffsetsExpired={}) in {} milliseconds.",
- groupId, records.size(), allOffsetsExpired, time.milliseconds() - startMs);
});
+ log.info("Generated {} tombstone records in {} milliseconds.", records.size(), time.milliseconds() - startMs);
Review Comment:
Should we add a little more context? We don't really know what this log is about when reading it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14870:
URL: https://github.com/apache/kafka/pull/14870#discussion_r1411843255
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -620,8 +621,9 @@ public boolean cleanupExpiredOffsets(String groupId, List<Record> records) {
});
if (!expiredPartitions.isEmpty()) {
- log.info("[GroupId {}] Expiring offsets of partitions (allOffsetsExpired={}): {}",
- groupId, allOffsetsExpired, String.join(", ", expiredPartitions));
+ log.info("[GroupId {}] Expiring {} offsets (allOffsetsExpired={}) in {} milliseconds.",
+ groupId, expiredPartitions.size(), allOffsetsExpired, time.milliseconds() - startMs);
+ log.debug("[GroupId {}] Expired partitions: {}", groupId, String.join(", ", expiredPartitions));
Review Comment:
Instead of this one, I wonder if we should just add a debug message for every expired offset at L613. What do you think?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/OffsetMetadataManager.java:
##########
@@ -620,8 +621,9 @@ public boolean cleanupExpiredOffsets(String groupId, List<Record> records) {
});
if (!expiredPartitions.isEmpty()) {
- log.info("[GroupId {}] Expiring offsets of partitions (allOffsetsExpired={}): {}",
- groupId, allOffsetsExpired, String.join(", ", expiredPartitions));
+ log.info("[GroupId {}] Expiring {} offsets (allOffsetsExpired={}) in {} milliseconds.",
Review Comment:
I don't really get the value of adding the time here. What's the reasoning behind it? However, I think that it would make sense to add a log in `cleanupGroupMetadata` which contains the time taken and the number of records generated.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorTimer.java:
##########
@@ -52,6 +52,20 @@ interface TimeoutOperation<T, U> {
*/
void schedule(String key, long delay, TimeUnit unit, boolean retry, TimeoutOperation<T, U> operation);
+ /**
+ * Add an operation to the timer. If an operation with the same key
+ * already exists, replace it with the new operation.
+ *
+ * @param key The key to identify this operation.
+ * @param delay The delay to wait before expiring.
+ * @param unit The delay unit.
+ * @param retry A boolean indicating whether the operation should
+ * be retried on failure.
+ * @param retryBackoff The delay when rescheduled on retry.
+ * @param operation The operation to perform upon expiration.
+ */
+ void schedule(String key, long delay, TimeUnit unit, boolean retry, long retryBackoff, TimeoutOperation<T, U> operation);
Review Comment:
Does it rely on the `unit`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java:
##########
@@ -281,6 +281,18 @@ public void schedule(
TimeUnit unit,
boolean retry,
TimeoutOperation<Void, U> operation
+ ) {
+ schedule(key, delay, unit, retry, 500, operation);
+ }
+
+ @Override
+ public void schedule(
+ String key,
+ long delay,
+ TimeUnit unit,
+ boolean retry,
+ long retryBackoff,
Review Comment:
What's the unit of this? Does it rely on `unit`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14987 [2/2]; customize retry backoff for group/offsets expiration [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14870:
URL: https://github.com/apache/kafka/pull/14870
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org