SKYEDEN-3271 | commiting offsets change on retransmission

allegro · Dec 23, 2024 · 9981d34 · 9981d34
1 parent aff7824
commit 9981d34
Show file tree

Hide file tree

Showing 12 changed files with 80 additions and 55 deletions.
diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/BatchConsumer.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/BatchConsumer.java
@@ -22,7 +22,7 @@
 import pl.allegro.tech.hermes.api.BatchSubscriptionPolicy;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.Topic;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.common.message.wrapper.CompositeMessageContentWrapper;
 import pl.allegro.tech.hermes.common.metric.MetricsFacade;
 import pl.allegro.tech.hermes.consumers.consumer.batch.MessageBatch;
@@ -239,11 +239,11 @@ public void commit(Set<SubscriptionPartitionOffset> offsetsToCommit) {
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset partitionOffset) {
+  public PartitionOffsets moveOffset(PartitionOffsets partitionOffsets) {
     if (receiver != null) {
-      return receiver.moveOffset(partitionOffset);
+      return receiver.moveOffset(partitionOffsets);
     }
-    return false;
+    return new PartitionOffsets();
   }
 
   @Override

diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/Consumer.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/Consumer.java
@@ -3,7 +3,7 @@
 import java.util.Set;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.Topic;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.consumers.consumer.offset.SubscriptionPartitionOffset;
 
 public interface Consumer {
@@ -26,7 +26,7 @@ public interface Consumer {
 
   void commit(Set<SubscriptionPartitionOffset> offsets);
 
-  boolean moveOffset(PartitionOffset subscriptionPartitionOffset);
+  PartitionOffsets moveOffset(PartitionOffsets subscriptionPartitionOffsets);
 
   Subscription getSubscription();
 }
diff --git a/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/SerialConsumer.java b/hermes-consumers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/SerialConsumer.java
@@ -11,7 +11,7 @@
 import org.slf4j.LoggerFactory;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.Topic;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.common.metric.MetricsFacade;
 import pl.allegro.tech.hermes.consumers.CommonConsumerParameters;
 import pl.allegro.tech.hermes.consumers.consumer.converter.MessageConverterResolver;
@@ -262,8 +262,8 @@ public void commit(Set<SubscriptionPartitionOffset> offsets) {
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset offset) {
-    return messageReceiver.moveOffset(offset);
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
+    return messageReceiver.moveOffset(offsets);
   }
 
   @Override

diff --git a/...s/src/main/java/pl/allegro/tech/hermes/consumers/consumer/batch/MessageBatchReceiver.java b/...s/src/main/java/pl/allegro/tech/hermes/consumers/consumer/batch/MessageBatchReceiver.java
@@ -16,7 +16,7 @@
 import org.slf4j.LoggerFactory;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.Topic;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.common.message.wrapper.CompositeMessageContentWrapper;
 import pl.allegro.tech.hermes.common.message.wrapper.UnsupportedContentTypeException;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
@@ -180,7 +180,7 @@ public void commit(Set<SubscriptionPartitionOffset> offsets) {
     receiver.commit(offsets);
   }
 
-  public boolean moveOffset(PartitionOffset offset) {
-    return receiver.moveOffset(offset);
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
+    return receiver.moveOffset(offsets);
   }
 }
diff --git a/.../allegro/tech/hermes/consumers/consumer/offset/kafka/broker/KafkaConsumerOffsetMover.java b/.../allegro/tech/hermes/consumers/consumer/offset/kafka/broker/KafkaConsumerOffsetMover.java
@@ -1,11 +1,15 @@
 package pl.allegro.tech.hermes.consumers.consumer.offset.kafka.broker;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import pl.allegro.tech.hermes.api.SubscriptionName;
 import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 
 public class KafkaConsumerOffsetMover {
 
@@ -19,7 +23,21 @@ public KafkaConsumerOffsetMover(SubscriptionName subscriptionName, KafkaConsumer
     this.consumer = consumer;
   }
 
-  public boolean move(PartitionOffset offset) {
+  public PartitionOffsets move(PartitionOffsets offsets) {
+    PartitionOffsets movedOffsets = new PartitionOffsets();
+
+    for (PartitionOffset offset : offsets) {
+      if (move(offset)) {
+        movedOffsets.add(offset);
+      }
+    }
+
+    commit(movedOffsets);
+
+    return movedOffsets;
+  }
+
+  private boolean move(PartitionOffset offset) {
     try {
       TopicPartition tp = new TopicPartition(offset.getTopic().asString(), offset.getPartition());
       if (consumer.assignment().contains(tp)) {
@@ -46,4 +64,24 @@ public boolean move(PartitionOffset offset) {
       return false;
     }
   }
+
+  private void commit(PartitionOffsets partitionOffsets) {
+    try {
+      Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new LinkedHashMap<>();
+      for (PartitionOffset partitionOffset : partitionOffsets) {
+        offsetsToCommit.put(
+            new TopicPartition(
+                partitionOffset.getTopic().asString(), partitionOffset.getPartition()),
+            new OffsetAndMetadata(partitionOffset.getOffset()));
+      }
+      if (!offsetsToCommit.isEmpty()) {
+        consumer.commitSync(offsetsToCommit);
+      }
+    } catch (Exception e) {
+      logger.error(
+          "Failed to commit offsets while trying to move them for subscription {}",
+          subscriptionName,
+          e);
+    }
+  }
 }
diff --git a/...ers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/receiver/MessageReceiver.java b/...ers/src/main/java/pl/allegro/tech/hermes/consumers/consumer/receiver/MessageReceiver.java
@@ -3,7 +3,7 @@
 import java.util.Optional;
 import java.util.Set;
 import pl.allegro.tech.hermes.api.Subscription;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
 import pl.allegro.tech.hermes.consumers.consumer.offset.SubscriptionPartitionOffset;
 
@@ -33,5 +33,5 @@ default void update(Subscription newSubscription) {}
 
   void commit(Set<SubscriptionPartitionOffset> offsets);
 
-  boolean moveOffset(PartitionOffset offset);
+  PartitionOffsets moveOffset(PartitionOffsets offsets);
 }
diff --git a/...in/java/pl/allegro/tech/hermes/consumers/consumer/receiver/ThrottlingMessageReceiver.java b/...in/java/pl/allegro/tech/hermes/consumers/consumer/receiver/ThrottlingMessageReceiver.java
@@ -4,7 +4,7 @@
 import java.util.Set;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.SubscriptionName;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.common.metric.MetricsFacade;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
 import pl.allegro.tech.hermes.consumers.consumer.idletime.IdleTimeCalculator;
@@ -53,8 +53,8 @@ public void commit(Set<SubscriptionPartitionOffset> offsets) {
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset offset) {
-    return receiver.moveOffset(offset);
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
+    return receiver.moveOffset(offsets);
   }
 
   @Override

diff --git a/...java/pl/allegro/tech/hermes/consumers/consumer/receiver/UninitializedMessageReceiver.java b/...java/pl/allegro/tech/hermes/consumers/consumer/receiver/UninitializedMessageReceiver.java
@@ -2,7 +2,7 @@
 
 import java.util.Optional;
 import java.util.Set;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
 import pl.allegro.tech.hermes.consumers.consumer.offset.SubscriptionPartitionOffset;
 
@@ -18,7 +18,7 @@ public void commit(Set<SubscriptionPartitionOffset> offsets) {
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset offset) {
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
     throw new ConsumerNotInitializedException();
   }
 }
diff --git a/...va/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/FilteringMessageReceiver.java b/...va/pl/allegro/tech/hermes/consumers/consumer/receiver/kafka/FilteringMessageReceiver.java
@@ -4,7 +4,7 @@
 import java.util.Optional;
 import java.util.Set;
 import pl.allegro.tech.hermes.api.Subscription;
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
 import pl.allegro.tech.hermes.consumers.consumer.filtering.FilteredMessageHandler;
 import pl.allegro.tech.hermes.consumers.consumer.offset.SubscriptionPartitionOffset;
@@ -65,7 +65,7 @@ public void commit(Set<SubscriptionPartitionOffset> offsets) {
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset offset) {
-    return receiver.moveOffset(offset);
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
+    return receiver.moveOffset(offsets);
   }
 }
diff --git a/...gro/tech/hermes/consumers/consumer/receiver/kafka/KafkaSingleThreadedMessageReceiver.java b/...gro/tech/hermes/consumers/consumer/receiver/kafka/KafkaSingleThreadedMessageReceiver.java
@@ -26,6 +26,7 @@
 import pl.allegro.tech.hermes.common.kafka.KafkaTopic;
 import pl.allegro.tech.hermes.common.kafka.KafkaTopics;
 import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset;
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets;
 import pl.allegro.tech.hermes.common.metric.MetricsFacade;
 import pl.allegro.tech.hermes.consumers.consumer.Message;
 import pl.allegro.tech.hermes.consumers.consumer.load.SubscriptionLoadRecorder;
@@ -223,7 +224,7 @@ private Map<TopicPartition, OffsetAndMetadata> createOffset(
   }
 
   @Override
-  public boolean moveOffset(PartitionOffset offset) {
-    return offsetMover.move(offset);
+  public PartitionOffsets moveOffset(PartitionOffsets offsets) {
+    return offsetMover.move(offsets);
   }
 }
diff --git a/...mers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/Retransmitter.java b/...mers/src/main/java/pl/allegro/tech/hermes/consumers/supervisor/process/Retransmitter.java
@@ -30,36 +30,22 @@ public void reloadOffsets(SubscriptionName subscriptionName, Consumer consumer)
           subscriptionOffsetChangeIndicator.getSubscriptionOffsets(
               subscriptionName.getTopicName(), subscriptionName.getName(), brokersClusterName);
 
-      for (PartitionOffset partitionOffset : offsets) {
-        if (moveOffset(subscriptionName, consumer, partitionOffset)) {
-          subscriptionOffsetChangeIndicator.removeOffset(
-              subscriptionName.getTopicName(),
-              subscriptionName.getName(),
-              brokersClusterName,
-              partitionOffset.getTopic(),
-              partitionOffset.getPartition());
-          logger.info(
-              "Removed offset indicator for subscription={} and partition={}",
-              subscriptionName,
-              partitionOffset.getPartition());
-        }
+      PartitionOffsets movedOffsets = consumer.moveOffset(offsets);
+
+      for (PartitionOffset partitionOffset : movedOffsets) {
+        subscriptionOffsetChangeIndicator.removeOffset(
+            subscriptionName.getTopicName(),
+            subscriptionName.getName(),
+            brokersClusterName,
+            partitionOffset.getTopic(),
+            partitionOffset.getPartition());
+        logger.info(
+            "Removed offset indicator for subscription={} and partition={}",
+            subscriptionName,
+            partitionOffset.getPartition());
       }
     } catch (Exception ex) {
       throw new RetransmissionException(ex);
     }
   }
-
-  private boolean moveOffset(
-      SubscriptionName subscriptionName, Consumer consumer, PartitionOffset partitionOffset) {
-    try {
-      return consumer.moveOffset(partitionOffset);
-    } catch (IllegalStateException ex) {
-      logger.warn(
-          "Cannot move offset for subscription={} and partition={} , possibly owned by different node",
-          subscriptionName,
-          partitionOffset.getPartition(),
-          ex);
-      return false;
-    }
-  }
 }
diff --git a/...s/src/test/groovy/pl/allegro/tech/hermes/consumers/supervisor/process/ConsumerStub.groovy b/...s/src/test/groovy/pl/allegro/tech/hermes/consumers/supervisor/process/ConsumerStub.groovy
@@ -2,7 +2,7 @@ package pl.allegro.tech.hermes.consumers.supervisor.process
 
 import pl.allegro.tech.hermes.api.Subscription
 import pl.allegro.tech.hermes.api.Topic
-import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffset
+import pl.allegro.tech.hermes.common.kafka.offset.PartitionOffsets
 import pl.allegro.tech.hermes.consumers.consumer.Consumer
 import pl.allegro.tech.hermes.consumers.consumer.offset.SubscriptionPartitionOffset
 
@@ -73,8 +73,8 @@ class ConsumerStub implements Consumer {
     }
 
     @Override
-    boolean moveOffset(PartitionOffset partitionOffset) {
-        return true
+    PartitionOffsets moveOffset(PartitionOffsets partitionOffset) {
+        return partitionOffset
     }
 
     boolean getInitialized() {