mimaison commented on code in PR #22324:
URL: https://github.com/apache/kafka/pull/22324#discussion_r3666987051
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -276,20 +282,66 @@ private void sendTopicStatus(final String connector,
final String topic, final T
final byte[] value = serializeTopicStatus(status);
+ sendWithRetry(key, value, 0);
+ }
+
+ /**
+ * Send a message to the status topic with retry logic using exponential
backoff.
+ *
+ * @param key the message key
+ * @param value the message value
+ * @param attemptNumber the current retry attempt number (0 for first
attempt)
+ */
+ private void sendWithRetry(final String key, final byte[] value, final int
attemptNumber) {
kafkaLog.send(key, value, new
org.apache.kafka.clients.producer.Callback() {
@Override
public void onCompletion(RecordMetadata metadata, Exception
exception) {
- if (exception == null) return;
- // TODO: retry more gracefully and not forever
+ if (exception == null) {
+ if (attemptNumber > 0) {
+ log.info("Successfully sent status update for key {}
after {} retry attempt(s)",
+ key, attemptNumber);
+ }
+ return;
+ }
+
if (exception instanceof RetriableException) {
- sendRetryExecutor.submit(() -> kafkaLog.send(key, value,
this));
+ long backoffMs = calculateBackoff(attemptNumber);
+ if (attemptNumber < BACKOFF_ESCALATION_THRESHOLD) {
+ log.warn("Failed to write status update for key {}
(attempt {}). " +
+ "Retrying after {}ms. Reason: {}",
+ key, attemptNumber + 1, backoffMs,
exception.getMessage());
+ } else {
+ log.warn("Failed to write status update for key {}
after {} attempts. " +
+ "Will continue retrying with {}ms backoff.
Reason: {}",
+ key, attemptNumber + 1, backoffMs,
exception.getMessage());
+ }
Review Comment:
I'm not sure having different messages is worth it.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -128,6 +128,12 @@ public class KafkaStatusBackingStore extends
KafkaTopicBasedBackingStore impleme
TOPIC_STATUS_VALUE_SCHEMA_V0
).build();
+ // Retry configuration constants
+ // After this many attempts, backoff will be capped at MAX_RETRY_BACKOFF_MS
+ private static final int BACKOFF_ESCALATION_THRESHOLD = 10;
+ private static final long INITIAL_RETRY_BACKOFF_MS = 300;
+ private static final long MAX_RETRY_BACKOFF_MS = 60000; // 60 seconds
Review Comment:
Rather than rolling a custom exponential backoff, could we reuse
`org.apache.kafka.common.utils.internals.ExponentialBackoff`? For example it's
already used in `DistributedHerder`.
--
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]
