HDDS-15562. EventNotification: persistence strategy for consumer checkpoint (via Ozone filesystem bucket properties)

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

@@ -0,0 +1,31 @@
+/*
+ * 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;
+}

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

@@ -35,4 +35,6 @@ public interface OMEventListenerPluginContext {
 
   // XXX: this probably doesn't belong here
   String getThreadNamePrefix();
+
+  NotificationCheckpointStrategy getNotificationCheckpointStrategy();
 }

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/OMEventListenerLedgerPoller.java

@@ -113,6 +113,10 @@ public int getPriority() {
     @Override
     public BackgroundTaskResult call() {
       if (shouldRun()) {
+        if (!seekPosition.verifyCheckpointAccess()) {
+          return BackgroundTaskResult.EmptyTaskResult.newResult();
+        }
+
         if (LOG.isDebugEnabled()) {
           LOG.debug("Running OMEventListenerLedgerPoller");
         }

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

@@ -17,41 +17,83 @@
 
 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 AtomicReference<String> seekPosition;
+  private final NotificationCheckpointStrategy checkpointStrategy;
+  private volatile boolean checkpointVerified = false;
 
-  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.error("Failed to load initial seek position from checkpoint strategy", ex);
+    }
     return null;
   }
 
   public String get() {
     return seekPosition.get();
   }
 
+  public boolean verifyCheckpointAccess() {
+    if (checkpointVerified) {
+      return true;
+    }
+    try {
+      if (checkpointStrategy != null) {
+        // Lightweight read-only check: loading from the strategy verifies that
+        // the underlying checkpoint volume and bucket exist and are accessible.
+        String loaded = checkpointStrategy.load();
+        if (seekPosition.get() == null) {
+          seekPosition.set(loaded);
+        }
+      }
+      checkpointVerified = true;
+      return true;
+    } catch (Exception ex) {
+      LOG.warn("Checkpoint storage is not accessible: {}", ex.getMessage());
+      return false;
+    }
+  }
+
   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)) {
+      checkpointVerified = true;
+      return;
+    }
+    try {
+      if (checkpointStrategy != null) {
+        checkpointStrategy.save(val);
+      }
+      // NOTE: this in-memory view of the seek position must only be kept
+      // up to date after we successfully persist it, so that any save
+      // failures prevent the poller from advancing and running away.
+      seekPosition.set(val);
+      checkpointVerified = true; // successful save confirms checkpoint is verified
+    } catch (Exception ex) {
+      checkpointVerified = false; // fail-safe: any save failure makes us unverified
+      LOG.error("Failed to save seek position checkpoint {}. Progress will not be advanced in-memory.", val, ex);
+    }
   }
 
   @Override

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

@@ -0,0 +1,187 @@
+/*
+ * 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.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+import java.util.Collections;
+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 {@link OMEventListenerLedgerPoller}'s failed-state and recovery tracking.
+ */
+@ExtendWith(MockitoExtension.class)
+public class TestOMEventListenerLedgerPollerRecovery {
+
+  @Mock
+  private OMEventListenerPluginContext pluginContext;
+
+  @Mock
+  private NotificationCheckpointStrategy checkpointStrategy;
+
+  @Mock
+  private Consumer<OmCompletedRequestInfo> callback;
+
+  @Test
+  public void testPollerSuspendsAndRecoversOnUnhealthyCheckpoint() throws Exception {
+    OzoneConfiguration conf = new OzoneConfiguration();
+
+    // 1. Initial successful startup / load (using safe doReturn stubbings)
+    doReturn("10").when(checkpointStrategy).load();
+    when(pluginContext.isLeaderReady()).thenReturn(true);
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    OMEventListenerLedgerPoller poller = new OMEventListenerLedgerPoller(
+        1000, TimeUnit.MILLISECONDS, 1, 5000,
+        pluginContext, conf, seekPosition, callback);
+
+    // Retrieve the background task
+    BackgroundTaskQueue queue = poller.getTasks();
+    BackgroundTask task = queue.poll();
+    Assertions.assertNotNull(task);
+
+    // Initial run: successful verify (which calls load()) and poll
+    when(pluginContext.listCompletedRequestInfo(eq(10L), anyInt()))
+        .thenReturn(Collections.emptyList());
+
+    task.call();
+
+    // Verification: load() was called 3 times in total:
+    // 1 in constructor, 1 in task first-run initSeekPosition(), and 1 in verifyCheckpointAccess()
+    verify(checkpointStrategy, times(3)).load();
+    verify(pluginContext, times(1)).listCompletedRequestInfo(eq(10L), anyInt());
+
+    // 2. Now simulate checkpoint strategy becoming unhealthy (save throws IOException)
+    // Make the last save fail by calling seekPosition.set("11") which we mock to fail
+    doThrow(new IOException("Save failed")).when(checkpointStrategy).save("11");
+    seekPosition.set("11");
+
+    // In-memory view remains "10"
+    Assertions.assertEquals("10", seekPosition.get());
+
+    // Now when task runs, verifyCheckpointAccess will be false because load() throws IOException
+    doThrow(new IOException("Storage missing")).when(checkpointStrategy).load();
+
+    task.call();
+
+    // Verify no additional calls to listCompletedRequestInfo were made (still only 1)
+    verify(pluginContext, times(1)).listCompletedRequestInfo(any(), anyInt());
+    verify(callback, never()).accept(any());
+
+    // 3. Simulate recovery of the checkpoint strategy (load() succeeds again)
+    doReturn("10").when(checkpointStrategy).load();
+
+    // Run the task again after recovery. It should resume polling.
+    task.call();
+
+    // Verification: listCompletedRequestInfo was called again (total 2)
+    verify(pluginContext, times(2)).listCompletedRequestInfo(eq(10L), anyInt());
+  }
+
+  @Test
+  public void testVerifyCheckpointAccess() throws Exception {
+    // 1. Mock load to throw exception during startup using safe doThrow stubbing
+    doThrow(new IOException("Failed to load")).when(checkpointStrategy).load();
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    // It should be unverified on startup
+    Assertions.assertFalse(seekPosition.verifyCheckpointAccess());
+
+    // 2. Mock load to succeed next using safe doReturn stubbing
+    doReturn("10").when(checkpointStrategy).load();
+    Assertions.assertTrue(seekPosition.verifyCheckpointAccess());
+    Assertions.assertEquals("10", seekPosition.get());
+
+    // Subsequent checks should be instant and not invoke load again (still only 3 total loads)
+    Assertions.assertTrue(seekPosition.verifyCheckpointAccess());
+    verify(checkpointStrategy, times(3)).load();
+  }
+
+  @Test
+  public void testSaveFailureUnverifiesCheckpoint() throws Exception {
+    doReturn("10").when(checkpointStrategy).load();
+
+    OMEventListenerLedgerPollerSeekPosition seekPosition =
+        new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy);
+
+    // Verifying is successful (calls load again)
+    Assertions.assertTrue(seekPosition.verifyCheckpointAccess());
+    verify(checkpointStrategy, times(2)).load();
+
+    // Successful save maintains verified state
+    doNothing().when(checkpointStrategy).save("11");
+    seekPosition.set("11");
+    Assertions.assertTrue(seekPosition.verifyCheckpointAccess());
+
+    // Failed save makes it unverified
+    doThrow(new IOException("Save failed")).when(checkpointStrategy).save("12");
+    seekPosition.set("12");
+
+    // In-memory stays "11"
+    Assertions.assertEquals("11", seekPosition.get());
+
+    // Must be unverified now, so verifyCheckpointAccess calls load() again.
+    // Let's mock load() to fail using safe doThrow.
+    doThrow(new IOException("Failed to verify")).when(checkpointStrategy).load();
+    Assertions.assertFalse(seekPosition.verifyCheckpointAccess());
+  }
+
+  @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");
+
+    // It should remain verified
+    Assertions.assertTrue(seekPosition.verifyCheckpointAccess());
+  }
+}

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

@@ -971,8 +971,6 @@ private void instantiateServices(boolean withNewSnapshot) throws IOException {
     prefixManager = new PrefixManagerImpl(this, metadataManager, true);
     keyManager = new KeyManagerImpl(this, scmClient, configuration,
         perfMetrics);
-    eventListenerPluginManager = new OMEventListenerPluginManager(this,
-        configuration);
     // If authorizer is not initialized or the authorizer is Native
     // re-initialize the authorizer, else for non-native authorizer
     // like ranger we can reuse previous value if it is initialized
@@ -996,6 +994,9 @@ public void close() {
       }
     };
 
+    eventListenerPluginManager = new OMEventListenerPluginManager(this,
+        configuration);
+
     // Reload snapshot feature config flag
     fsSnapshotEnabled = configuration.getBoolean(
         OMConfigKeys.OZONE_FILESYSTEM_SNAPSHOT_ENABLED_KEY,

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
@@ -50,4 +53,9 @@ public List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int
   public String getThreadNamePrefix() {
     return ozoneManager.getThreadNamePrefix();
   }
+
+  @Override
+  public NotificationCheckpointStrategy getNotificationCheckpointStrategy() {
+    return checkpointStrategy;
+  }
 }

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 OzoneFileCheckpointStrategy(ozoneManager, conf);
+    }
+    OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager, checkpointStrategy);
 
     for (String destName : destNameList) {
       try {