mimaison commented on code in PR #22324:
URL: https://github.com/apache/kafka/pull/22324#discussion_r3435875285
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -276,20 +281,64 @@ 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));
+ if (attemptNumber >= MAX_RETRY_ATTEMPTS) {
+ log.error("Failed to write status update for key {}
after {} attempts. Giving up.",
+ key, attemptNumber + 1, exception);
+ return;
+ }
Review Comment:
I'm uneasy with completely giving up and only logging it, especially on a
`RetriableException`
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -276,20 +281,64 @@ 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));
+ if (attemptNumber >= MAX_RETRY_ATTEMPTS) {
+ log.error("Failed to write status update for key {}
after {} attempts. Giving up.",
+ key, attemptNumber + 1, exception);
+ return;
+ }
+
+ long backoffMs = calculateBackoff(attemptNumber);
+ log.warn("Failed to write status update for key {}
(attempt {}/{}). " +
+ "Retrying after {}ms. Reason: {}",
+ key, attemptNumber + 1, MAX_RETRY_ATTEMPTS + 1,
backoffMs, exception.getMessage());
+
+ sendRetryExecutor.submit(() -> {
+ time.sleep(backoffMs);
Review Comment:
Shouldn't we use a ScheduledExecutor instead of doing `sleep()`?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -276,20 +281,64 @@ 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));
+ if (attemptNumber >= MAX_RETRY_ATTEMPTS) {
+ log.error("Failed to write status update for key {}
after {} attempts. Giving up.",
+ key, attemptNumber + 1, exception);
+ return;
+ }
+
+ long backoffMs = calculateBackoff(attemptNumber);
+ log.warn("Failed to write status update for key {}
(attempt {}/{}). " +
+ "Retrying after {}ms. Reason: {}",
+ key, attemptNumber + 1, MAX_RETRY_ATTEMPTS + 1,
backoffMs, exception.getMessage());
Review Comment:
The message does not seem correct. The first attempt will print 1/11 and the
last one will print 10/11, shouldn't it be 1/10 to 10/10?
--
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]
