HDDS-15562. EventNotification: persistence strategy for consumer checkpoint (via distributed RocksDB update)

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -133,6 +133,9 @@ public final class OzoneConsts {
   public static final String RANGER_OZONE_SERVICE_VERSION_KEY =
       "#RANGEROZONESERVICEVERSION";
 
+  public static final String EVENT_NOTIFICATION_CHECKPOINT_PREFIX =
+      "#EVENTNOTIFICATIONCHECKPOINT#";
+
   public static final String MULTIPART_FORM_DATA_BOUNDARY = "---XXX";
 
   // Block ID prefixes used in datanode containers.

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java

@@ -307,6 +307,7 @@ public static boolean isReadOnly(OMRequest omRequest) {
     case TenantAssignAdmin:
     case TenantRevokeAdmin:
     case SetRangerServiceVersion:
+    case SetEventNotificationCheckpoint:
     case CreateSnapshot:
     case DeleteSnapshot:
     case RenameSnapshot:
@@ -422,6 +423,7 @@ public static boolean shouldSendToFollower(OMRequest omRequest) {
     case TenantAssignAdmin:
     case TenantRevokeAdmin:
     case SetRangerServiceVersion:
+    case SetEventNotificationCheckpoint:
     case CreateSnapshot:
     case DeleteSnapshot:
     case RenameSnapshot:

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java

@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.eventlistener;
+
+import java.io.IOException;
+
+/**
+ * Interface for implementations which load/save the current checkpoint
+ * transaction log index used by an event poller.
+ */
+public interface NotificationCheckpointStrategy {
+
+  String load() throws IOException;
+
+  void save(String val) throws IOException;
+
+  void reset() throws IOException;
+}

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java

@@ -33,6 +33,8 @@ public interface OMEventListenerPluginContext {
   // them to some predefined value for safety?  e.g. 10K
   List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int maxResults) throws IOException;
 
+  NotificationCheckpointStrategy getNotificationCheckpointStrategy();
+
   // XXX: this probably doesn't belong here
   String getThreadNamePrefix();
 }

hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto

@@ -156,6 +156,7 @@ enum Type {
   PutObjectTagging = 140;
   GetObjectTagging = 141;
   DeleteObjectTagging = 142;
+  SetEventNotificationCheckpoint = 144;
 }
 
 enum SafeMode {
@@ -304,6 +305,7 @@ message OMRequest {
   optional PutObjectTaggingRequest          putObjectTaggingRequest        = 141;
   optional DeleteObjectTaggingRequest       deleteObjectTaggingRequest     = 142;
   repeated SetSnapshotPropertyRequest       SetSnapshotPropertyRequests    = 143;
+  optional SetEventNotificationCheckpointRequest setEventNotificationCheckpointRequest = 144;
 }
 
 message OMResponse {
@@ -437,6 +439,7 @@ message OMResponse {
   optional GetObjectTaggingResponse          getObjectTaggingResponse      = 140;
   optional PutObjectTaggingResponse          putObjectTaggingResponse      = 141;
   optional DeleteObjectTaggingResponse       deleteObjectTaggingResponse   = 142;
+  optional SetEventNotificationCheckpointResponse setEventNotificationCheckpointResponse = 144;
 }
 
 enum Status {
@@ -2003,6 +2006,11 @@ message SetRangerServiceVersionRequest {
     required uint64 rangerServiceVersion = 1;
 }
 
+message SetEventNotificationCheckpointRequest {
+    required string checkpointKey = 1;
+    required string checkpointValue = 2;
+}
+
 message CreateSnapshotRequest {
   optional string volumeName = 1;
   optional string bucketName = 2;
@@ -2156,6 +2164,9 @@ message CreateTenantResponse {
 message SetRangerServiceVersionResponse {
 }
 
+message SetEventNotificationCheckpointResponse {
+}
+
 message CreateSnapshotResponse {
   optional SnapshotInfo snapshotInfo = 1;
 }

hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java

@@ -84,7 +84,8 @@ public void initialize(OzoneConfiguration conf, OMEventListenerPluginContext plu
       exception.initCause(ex);
       throw exception;
     }
-    this.seekPosition = new OMEventListenerLedgerPollerSeekPosition();
+    this.seekPosition = new OMEventListenerLedgerPollerSeekPosition(
+        pluginContext.getNotificationCheckpointStrategy());
 
     LOG.info("Creating OMEventListenerLedgerPoller with serviceInterval={}," +
         "serviceTimeout={}, seekPosition={}",

hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java

@@ -17,28 +17,35 @@
 
 package org.apache.hadoop.ozone.om.eventlistener;
 
+import java.util.Objects;
 import java.util.concurrent.atomic.AtomicReference;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
  * This is a helper class to get/set the seek position used by the
  * OMEventListenerLedgerPoller.
- *
- * XXX: the seek position should be persisted (and ideally distributed to
- * all OMs) but at the moment it only lives in memory
  */
 public class OMEventListenerLedgerPollerSeekPosition {
   public static final Logger LOG = LoggerFactory.getLogger(OMEventListenerLedgerPollerSeekPosition.class);
 
+  private final NotificationCheckpointStrategy checkpointStrategy;
   private final AtomicReference<String> seekPosition;
 
-  public OMEventListenerLedgerPollerSeekPosition() {
-    this.seekPosition = new AtomicReference(initSeekPosition());
+  public OMEventListenerLedgerPollerSeekPosition(NotificationCheckpointStrategy checkpointStrategy) {
+    this.checkpointStrategy = checkpointStrategy;
+    this.seekPosition = new AtomicReference<>(initSeekPosition());
   }
 
-  // TODO: load this from persistent storage
   public String initSeekPosition() {
+    try {
+      if (checkpointStrategy != null) {
+        return checkpointStrategy.load();
+      }
+    } catch (Exception ex) {
+      LOG.warn("Failed to load initial seek position from checkpoint strategy. " +
+          "Fallback to starting from the beginning.", ex);
+    }
     return null;
   }
 
@@ -48,10 +55,36 @@ public String get() {
 
   public void set(String val) {
     LOG.debug("Setting seek position {}", val);
-    // NOTE: this in-memory view of the seek position needs to be kept
-    // up to date because the OMEventListenerLedgerPoller has a
-    // reference to it
-    seekPosition.set(val);
+    String current = seekPosition.get();
+    if (Objects.equals(current, val)) {
+      return;
+    }
+    try {
+      if (checkpointStrategy != null) {
+        checkpointStrategy.save(val);
+      }
+    } catch (Exception ex) {
+      // Save failure does NOT block subsequent runs or in-memory progress!
+      LOG.warn("Failed to save seek position checkpoint {} to database. " +
+          "Progress will continue in-memory but is not durably saved.", val, ex);
+    } finally {
+      // ALWAYS advance the in-memory seek position, even if saving fails,
+      // so that the background task is not blocked from making progress.
+      seekPosition.set(val);
+    }
+  }
+
+  public void reset() {
+    LOG.debug("Resetting seek position");
+    try {
+      if (checkpointStrategy != null) {
+        checkpointStrategy.reset();
+      }
+    } catch (Exception ex) {
+      LOG.warn("Failed to reset seek position checkpoint on database strategy.", ex);
+    } finally {
+      seekPosition.set(null);
+    }
   }
 
   @Override

hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java

@@ -0,0 +1,121 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.ozone.om.eventlistener;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.utils.BackgroundTask;
+import org.apache.hadoop.hdds.utils.BackgroundTaskQueue;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+/**
+ * Tests for OMEventListenerLedgerPoller recovery and checkpoint non-blocking behavior.
+ */
+@ExtendWith(MockitoExtension.class)
+public class TestOMEventListenerLedgerPollerRecovery {
+
+  @Mock
+  private OMEventListenerPluginContext pluginContext;
+
+  @Mock
+  private NotificationCheckpointStrategy checkpointStrategy;
+
+  @Mock
+  private Consumer<OmCompletedRequestInfo> callback;
+
+  @Test
+  public void testSaveFailureIsCompletelyNonBlocking() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+
+    when(pluginContext.isLeaderReady()).thenReturn(true);
+    when(pluginContext.getThreadNamePrefix()).thenReturn("test-poller-");
+
+    // Mock load to return 10
+    doReturn("10").when(checkpointStrategy).load();
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    // Initial position in memory should be 10
+    Assertions.assertEquals("10", seekPosition.get());
+
+    OMEventListenerLedgerPoller poller = new OMEventListenerLedgerPoller(
+        1000, TimeUnit.MILLISECONDS, 1, 1000,
+        pluginContext, conf, seekPosition, callback);
+
+    BackgroundTaskQueue queue = poller.getTasks();
+    BackgroundTask task = queue.poll();
+
+    // 1. First poller run: succeeds
+    task.call();
+    verify(pluginContext).listCompletedRequestInfo(any(), anyInt());
+
+    // 2. Now simulate checkpoint strategy becoming unhealthy (save throws IOException)
+    doThrow(new IOException("Save failed")).when(checkpointStrategy).save("11");
+    seekPosition.set("11");
+
+    // In-memory view should STILL be advanced to "11" so we don't block progress
+    Assertions.assertEquals("11", seekPosition.get());
+  }
+
+  @Test
+  public void testLoadFailureOnStartupFallbackToNull() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+
+    // Mock load to fail during initialization / startup
+    doThrow(new IOException("Database down")).when(checkpointStrategy).load();
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    // Initial load failure means we fallback to starting from the beginning (null)
+    Assertions.assertNull(seekPosition.get());
+  }
+
+  @Test
+  public void testSetToSameValueDoesNotTriggerSave() throws Exception {
+    doReturn("10").when(checkpointStrategy).load();
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    // Initial position in memory should be "10"
+    Assertions.assertEquals("10", seekPosition.get());
+
+    // Setting to the same value "10" should skip calling save on checkpointStrategy
+    seekPosition.set("10");
+
+    // Verify checkpointStrategy.save("10") was never called
+    verify(checkpointStrategy, never()).save("10");
+  }
+}

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java

@@ -28,9 +28,12 @@
  */
 public final class OMEventListenerPluginContextImpl implements OMEventListenerPluginContext {
   private final OzoneManager ozoneManager;
+  private final NotificationCheckpointStrategy checkpointStrategy;
 
-  public OMEventListenerPluginContextImpl(OzoneManager ozoneManager) {
+  public OMEventListenerPluginContextImpl(OzoneManager ozoneManager,
+                                          NotificationCheckpointStrategy checkpointStrategy) {
     this.ozoneManager = ozoneManager;
+    this.checkpointStrategy = checkpointStrategy;
   }
 
   @Override
@@ -45,6 +48,11 @@ public List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int
     return ozoneManager.getMetadataManager().listCompletedRequestInfo(startKey, maxResults);
   }
 
+  @Override
+  public NotificationCheckpointStrategy getNotificationCheckpointStrategy() {
+    return checkpointStrategy;
+  }
+
   // TODO: it feels like this doesn't belong here
   @Override
   public String getThreadNamePrefix() {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java

@@ -92,7 +92,11 @@ static List<OMEventListener> loadAll(OzoneManager ozoneManager, OzoneConfigurati
       }
     }
 
-    OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager);
+    NotificationCheckpointStrategy checkpointStrategy = null;
+    if (ozoneManager != null) {
+      checkpointStrategy = new OzoneDbCheckpointStrategy(ozoneManager, conf);
+    }
+    OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager, checkpointStrategy);
 
     for (String destName : destNameList) {
       try {