akhil-sabu-ibm commented on code in PR #22324:
URL: https://github.com/apache/kafka/pull/22324#discussion_r3538464663
##########
connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java:
##########
@@ -276,20 +281,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 < MAX_RETRY_ATTEMPTS) {
Review Comment:
@mimaison The PR has been updated with the changes. Could you please review?
--
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]
