From 3c97d0ef01ce384356d7fb5b4cf013f1c8e487cc Mon Sep 17 00:00:00 2001 From: gardenia Date: Fri, 12 Jun 2026 13:50:17 +0100 Subject: [PATCH 1/3] HDDS-15562. EventNotification: persist checkpoint Persist the checkpoint within the Ozone filesystem itself so that another OM can pick up if the leader changes. --- .../NotificationCheckpointStrategy.java | 31 ++++ .../OMEventListenerPluginContext.java | 2 + .../OMEventListenerKafkaPublisher.java | 3 +- ...EventListenerLedgerPollerSeekPosition.java | 32 +++- .../apache/hadoop/ozone/om/OzoneManager.java | 5 +- .../OMEventListenerPluginContextImpl.java | 10 +- .../OMEventListenerPluginManager.java | 7 +- .../OzoneFileCheckpointStrategy.java | 174 ++++++++++++++++++ .../TestOzoneFileCheckpointStrategy.java | 115 ++++++++++++ 9 files changed, 364 insertions(+), 15 deletions(-) create mode 100644 hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java create mode 100644 hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java create mode 100644 hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java new file mode 100644 index 000000000000..b91966462ce8 --- /dev/null +++ b/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; +} diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java index f61b5efa6120..996ae937ab4d 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java +++ b/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(); } diff --git a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java index f42524ba98ad..96513188c731 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java +++ b/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={}", diff --git a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java index bccbda1e2a7e..2c35dcc82f18 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java +++ b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java @@ -24,21 +24,26 @@ /** * 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 seekPosition; + private final NotificationCheckpointStrategy checkpointStrategy; - 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; } @@ -48,10 +53,17 @@ 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); + 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); + } catch (Exception ex) { + LOG.error("Failed to save seek position checkpoint {}. Progress will not be advanced in-memory.", val, ex); + } } @Override diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java index 93a601f7d08f..7313aba9b9b7 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java +++ b/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, diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java index e0a805f0afc7..7336dcb046e9 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java +++ b/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 listCompletedRequestInfo(Long startKey, int public String getThreadNamePrefix() { return ozoneManager.getThreadNamePrefix(); } + + @Override + public NotificationCheckpointStrategy getNotificationCheckpointStrategy() { + return checkpointStrategy; + } } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java index 79674dc20ff8..6e5b6c992866 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java @@ -92,7 +92,12 @@ static List loadAll(OzoneManager ozoneManager, OzoneConfigurati } } - OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager); + NotificationCheckpointStrategy checkpointStrategy = null; + if (ozoneManager != null && ozoneManager.getOmMetadataReader() != null) { + checkpointStrategy = new OzoneFileCheckpointStrategy(ozoneManager, + ozoneManager.getOmMetadataReader().get(), conf); + } + OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager, checkpointStrategy); for (String destName : destNameList) { try { diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java new file mode 100644 index 000000000000..da8c387dd31b --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java @@ -0,0 +1,174 @@ +/* + * 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 com.google.protobuf.ServiceException; +import java.io.IOException; +import java.net.InetAddress; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.commons.lang3.StringUtils; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.proto.HddsProtos; +import org.apache.hadoop.ozone.om.IOmMetadataReader; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; +import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketArgs; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetBucketPropertyRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo; +import org.apache.hadoop.security.UserGroupInformation; +import org.apache.ratis.protocol.ClientId; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * An implementation of NotificationCheckpointStrategy which loads/saves + * the the last known notification sent by an event notification plugin + * directly as metadata on the checkpoint bucket itself. + * + * This allows another OM to pick up from the appropriate place in the + * event of a leadership change. + * + * NOTE: The current approach is to store the current checkpoint as a + * bucket metadata property. This has the virtue of requiring only a + * single request (the first implementation of this used a two-phase + * CreateKeyRequest / CommitKeyRequest approach with the checkpoint + * value being stored as a KeyArgs metadata property, but the two-phase + * nature increased complexity). + */ +public class OzoneFileCheckpointStrategy implements NotificationCheckpointStrategy { + + public static final Logger LOG = LoggerFactory.getLogger(OzoneFileCheckpointStrategy.class); + + public static final String OZONE_OM_PLUGIN_CHECKPOINT_VOLUME = "ozone.om.plugin.kafka.checkpoint.volume"; + public static final String OZONE_OM_PLUGIN_CHECKPOINT_VOLUME_DEFAULT = "notifications"; + + public static final String OZONE_OM_PLUGIN_CHECKPOINT_BUCKET = "ozone.om.plugin.kafka.checkpoint.bucket"; + public static final String OZONE_OM_PLUGIN_CHECKPOINT_BUCKET_DEFAULT = "checkpoint"; + + public static final String OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL = + "ozone.om.plugin.kafka.checkpoint.save.interval"; + public static final int OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL_DEFAULT = 100; + + private static final String METDATA_KEY = "notification-checkpoint"; + + private final AtomicLong callId = new AtomicLong(0); + private final ClientId clientId = ClientId.randomId(); + private final OzoneManager ozoneManager; + private final AtomicLong saveCount = new AtomicLong(0); + + private final String volume; + private final String bucket; + private final int saveInterval; + + public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, final IOmMetadataReader omMetadataReader, + OzoneConfiguration conf) { + this.ozoneManager = ozoneManager; + this.volume = conf.get(OZONE_OM_PLUGIN_CHECKPOINT_VOLUME, OZONE_OM_PLUGIN_CHECKPOINT_VOLUME_DEFAULT); + this.bucket = conf.get(OZONE_OM_PLUGIN_CHECKPOINT_BUCKET, OZONE_OM_PLUGIN_CHECKPOINT_BUCKET_DEFAULT); + + int interval = conf.getInt(OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL, + OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL_DEFAULT); + if (interval < 1) { + LOG.warn("Configured save interval {} is invalid. Defaulting to 100.", interval); + interval = OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL_DEFAULT; + } + this.saveInterval = interval; + } + + @Override + public String load() throws IOException { + try { + OmBucketInfo bucketInfo = ozoneManager.getBucketInfo(volume, bucket); + if (bucketInfo != null && bucketInfo.getMetadata() != null) { + return bucketInfo.getMetadata().get(METDATA_KEY); + } + } catch (IOException ex) { + LOG.info("Error loading notification checkpoint from bucket /{}/{} - {}", volume, bucket, ex.getMessage()); + } + return null; + } + + @Override + public void save(String val) throws IOException { + if (StringUtils.isBlank(val)) { + return; + } + long previousSaveCount = saveCount.getAndIncrement(); + // Throttle database commits: persist checkpoint based on configured interval to avoid write storms + if (previousSaveCount == 0 || previousSaveCount % saveInterval == 0) { + saveImpl(val); + } + } + + private void saveImpl(String val) { + BucketArgs bucketArgs = BucketArgs.newBuilder() + .setVolumeName(volume) + .setBucketName(bucket) + .addMetadata(HddsProtos.KeyValue.newBuilder() + .setKey(METDATA_KEY) + .setValue(val) + .build()) + .build(); + + SetBucketPropertyRequest setBucketPropertyRequest = SetBucketPropertyRequest.newBuilder() + .setBucketArgs(bucketArgs) + .build(); + + OMRequest omRequest = OMRequest.newBuilder() + .setCmdType(Type.SetBucketProperty) + .setClientId(clientId.toString()) + .setSetBucketPropertyRequest(setBucketPropertyRequest) + .setUserInfo(getUserInfo()) + .build(); + + submitRequest(omRequest); + LOG.info("Persisted {} = {} directly as metadata on bucket /{}/{}", METDATA_KEY, val, volume, bucket); + } + + private UserInfo getUserInfo() { + UserInfo.Builder userInfo = UserInfo.newBuilder(); + try { + userInfo.setUserName(UserGroupInformation.getCurrentUser().getShortUserName()); + } catch (IOException e) { + LOG.warn("Failed to get current login user name", e); + userInfo.setUserName("om"); + } + + if (ozoneManager.getOmRpcServerAddr() != null) { + InetAddress remoteAddress = ozoneManager.getOmRpcServerAddr().getAddress(); + if (remoteAddress != null) { + userInfo.setHostName(remoteAddress.getHostName()); + userInfo.setRemoteAddress(remoteAddress.getHostAddress()); + } + } + return userInfo.build(); + } + + private OMResponse submitRequest(OMRequest omRequest) { + try { + return OzoneManagerRatisUtils.submitRequest(ozoneManager, omRequest, clientId, callId.incrementAndGet()); + } catch (ServiceException e) { + LOG.error("Set bucket metadata " + omRequest.getCmdType() + " request failed. Will retry at next run.", e); + } + return null; + } +} diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java new file mode 100644 index 000000000000..9a7c92d2978f --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java @@ -0,0 +1,115 @@ +/* + * 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.Mockito.mockStatic; +import static org.mockito.Mockito.when; + +import com.google.protobuf.ServiceException; +import java.io.IOException; +import org.apache.hadoop.ozone.om.IOmMetadataReader; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; +import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos; +import org.apache.ratis.protocol.ClientId; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.Mockito; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Tests {@link OzoneFileCheckpointStrategy}. + */ +@ExtendWith(MockitoExtension.class) +public class TestOzoneFileCheckpointStrategy { + + @Mock + private OzoneManager mockOzoneManager; + @Mock + private IOmMetadataReader mockOmMetadataReader; + @Mock + private OzoneManagerProtocolProtos.OMResponse mockOmResponse; + @Mock + private OmBucketInfo mockBucketInfo; + + private OzoneFileCheckpointStrategy ozoneFileCheckpointStrategy; + + @BeforeEach + public void setup() { + ozoneFileCheckpointStrategy = new OzoneFileCheckpointStrategy(mockOzoneManager, mockOmMetadataReader, + new org.apache.hadoop.hdds.conf.OzoneConfiguration()); + } + + @Test + public void testSaveStrategy() throws IOException, ServiceException { + try (MockedStatic utils = mockStatic(OzoneManagerRatisUtils.class)) { + utils.when(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class))).thenReturn(mockOmResponse); + + // Check its saved on first iteration + ozoneFileCheckpointStrategy.save("00000000000000000001"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + // But not on second + ozoneFileCheckpointStrategy.save("0000000000000000002"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + for (int i = 0; i <= 100; i++) { + String val = String.format("%020d", i); + ozoneFileCheckpointStrategy.save(val); + } + + // Check submit has only ran twice in total (first save + save at 100th iteration) + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(2)); + } + } + + @Test + public void testLoadStrategyWhenMetadataNotSet() throws IOException { + when(mockOzoneManager.getBucketInfo(any(), any())).thenReturn(mockBucketInfo); + when(mockBucketInfo.getMetadata()).thenReturn(com.google.common.collect.ImmutableMap.of()); + Assertions.assertNull(ozoneFileCheckpointStrategy.load()); + } + + @Test + public void testLoadStrategyWhenBucketDoesNotExist() throws IOException { + when(mockOzoneManager.getBucketInfo(any(), any())).thenThrow(IOException.class); + Assertions.assertNull(ozoneFileCheckpointStrategy.load()); + } + + @Test + public void testLoadStrategyWithValidMetaData() throws IOException { + when(mockOzoneManager.getBucketInfo(any(), any())).thenReturn(mockBucketInfo); + when(mockBucketInfo.getMetadata()).thenReturn( + com.google.common.collect.ImmutableMap.of("notification-checkpoint", "00000000000000000017")); + Assertions.assertEquals("00000000000000000017", ozoneFileCheckpointStrategy.load()); + } +} From d54d277e3ae967c736e3555317304ad74ddb8430 Mon Sep 17 00:00:00 2001 From: Colm Dougan Date: Tue, 16 Jun 2026 16:02:37 +0100 Subject: [PATCH 2/3] HDDS-15562. EventNotification: hardened for error cases If the event notification checkpoint bucket does not exist (or becomes inaccessible due to permissions/disk full), SetBucketProperty OMRequests fail at the state machine level. Previously, these failures were caught but ignored, allowing the in-memory seek position to advance while no checkpoint was written. Upon OM restart, the poller loaded null and replayed all events starting from transaction 1, creating infinite duplicate event storms on Kafka. Summary of improvements: * Before polling or publishing any events, verify checkpoint storage exists via a lightweight, read-only load() call. If it fails, the poller logs a wanring, suspends any Kafka publishes, and waits to retry on the next 2-second poll. * In the OzoneFileCheckpointStrategy we propagate IOExceptions when SetBucketProperty OMRequests return a non-OK status. Inside seekPosition.set(), catching this exception sets checkpointVerified to false and blocks in-memory checkpoint advancement. This suspends any Kafka publishes on subsequent polls. * Reset saveCount to 0 upon catching a write IOException. Once the storage becomes accessible (read succeeds), the first subsequent save bypasses throttling (writes immediately) to confirm write-path accessibility before resuming normal throttled writes. --- .../OMEventListenerLedgerPoller.java | 4 + ...EventListenerLedgerPollerSeekPosition.java | 24 +++ ...stOMEventListenerLedgerPollerRecovery.java | 167 ++++++++++++++++++ .../OzoneFileCheckpointStrategy.java | 38 ++-- .../TestOzoneFileCheckpointStrategy.java | 48 ++++- 5 files changed, 266 insertions(+), 15 deletions(-) create mode 100644 hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java diff --git a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPoller.java b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPoller.java index 2ae3a2ff70cc..8364a2b95b5c 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPoller.java +++ b/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"); } diff --git a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java index 2c35dcc82f18..8fe4107a31dc 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java +++ b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java @@ -30,6 +30,7 @@ public class OMEventListenerLedgerPollerSeekPosition { private final AtomicReference seekPosition; private final NotificationCheckpointStrategy checkpointStrategy; + private volatile boolean checkpointVerified = false; public OMEventListenerLedgerPollerSeekPosition(NotificationCheckpointStrategy checkpointStrategy) { this.checkpointStrategy = checkpointStrategy; @@ -51,6 +52,27 @@ 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); try { @@ -61,7 +83,9 @@ public void set(String val) { // 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); } } diff --git a/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java new file mode 100644 index 000000000000..177638afc9fb --- /dev/null +++ b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java @@ -0,0 +1,167 @@ +/* + * 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 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()); + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java index da8c387dd31b..9360d64df784 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java @@ -32,6 +32,7 @@ import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetBucketPropertyRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Status; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type; import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.UserInfo; import org.apache.hadoop.security.UserGroupInformation; @@ -96,13 +97,9 @@ public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, final IOmMetadataR @Override public String load() throws IOException { - try { - OmBucketInfo bucketInfo = ozoneManager.getBucketInfo(volume, bucket); - if (bucketInfo != null && bucketInfo.getMetadata() != null) { - return bucketInfo.getMetadata().get(METDATA_KEY); - } - } catch (IOException ex) { - LOG.info("Error loading notification checkpoint from bucket /{}/{} - {}", volume, bucket, ex.getMessage()); + OmBucketInfo bucketInfo = ozoneManager.getBucketInfo(volume, bucket); + if (bucketInfo != null && bucketInfo.getMetadata() != null) { + return bucketInfo.getMetadata().get(METDATA_KEY); } return null; } @@ -113,13 +110,20 @@ public void save(String val) throws IOException { return; } long previousSaveCount = saveCount.getAndIncrement(); - // Throttle database commits: persist checkpoint based on configured interval to avoid write storms + // Throttle database commits: persist checkpoint based on configured interval to avoid write storms. if (previousSaveCount == 0 || previousSaveCount % saveInterval == 0) { - saveImpl(val); + try { + saveImpl(val); + } catch (IOException e) { + // If save fails, reset saveCount to 0 so that the next save attempt + // will bypass throttling and try to write immediately upon recovery. + saveCount.set(0); + throw e; + } } } - private void saveImpl(String val) { + private void saveImpl(String val) throws IOException { BucketArgs bucketArgs = BucketArgs.newBuilder() .setVolumeName(volume) .setBucketName(bucket) @@ -163,12 +167,18 @@ private UserInfo getUserInfo() { return userInfo.build(); } - private OMResponse submitRequest(OMRequest omRequest) { + private OMResponse submitRequest(OMRequest omRequest) throws IOException { try { - return OzoneManagerRatisUtils.submitRequest(ozoneManager, omRequest, clientId, callId.incrementAndGet()); + OMResponse response = OzoneManagerRatisUtils.submitRequest( + ozoneManager, omRequest, clientId, callId.incrementAndGet()); + if (response != null && response.getStatus() != Status.OK) { + throw new IOException("Failed to persist checkpoint: " + response.getStatus() + + (StringUtils.isNotBlank(response.getMessage()) ? " - " + response.getMessage() : "")); + } + return response; } catch (ServiceException e) { - LOG.error("Set bucket metadata " + omRequest.getCmdType() + " request failed. Will retry at next run.", e); + LOG.error("Set bucket metadata " + omRequest.getCmdType() + " request failed.", e); + throw new IOException(e); } - return null; } } diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java index 9a7c92d2978f..d83cc53b6c64 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java @@ -67,6 +67,7 @@ public void testSaveStrategy() throws IOException, ServiceException { utils.when(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), any(OzoneManagerProtocolProtos.OMRequest.class), any(ClientId.class), any(Long.class))).thenReturn(mockOmResponse); + when(mockOmResponse.getStatus()).thenReturn(OzoneManagerProtocolProtos.Status.OK); // Check its saved on first iteration ozoneFileCheckpointStrategy.save("00000000000000000001"); @@ -102,7 +103,7 @@ public void testLoadStrategyWhenMetadataNotSet() throws IOException { @Test public void testLoadStrategyWhenBucketDoesNotExist() throws IOException { when(mockOzoneManager.getBucketInfo(any(), any())).thenThrow(IOException.class); - Assertions.assertNull(ozoneFileCheckpointStrategy.load()); + Assertions.assertThrows(IOException.class, () -> ozoneFileCheckpointStrategy.load()); } @Test @@ -112,4 +113,49 @@ public void testLoadStrategyWithValidMetaData() throws IOException { com.google.common.collect.ImmutableMap.of("notification-checkpoint", "00000000000000000017")); Assertions.assertEquals("00000000000000000017", ozoneFileCheckpointStrategy.load()); } + + @Test + public void testSaveStrategyBypassesThrottlingUponFailureRecovery() throws IOException, ServiceException { + try (MockedStatic utils = mockStatic(OzoneManagerRatisUtils.class)) { + utils.when(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class))).thenReturn(mockOmResponse); + + // First save succeeds (saveCount becomes 1) + when(mockOmResponse.getStatus()).thenReturn(OzoneManagerProtocolProtos.Status.OK); + ozoneFileCheckpointStrategy.save("1"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + // Saves 2 to 100 are throttled + for (int i = 2; i <= 100; i++) { + ozoneFileCheckpointStrategy.save(String.valueOf(i)); + } + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + // Save 101 tries to write and fails! + when(mockOmResponse.getStatus()).thenReturn(OzoneManagerProtocolProtos.Status.BUCKET_NOT_FOUND); + Assertions.assertThrows(IOException.class, () -> ozoneFileCheckpointStrategy.save("101")); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(2)); + + // Save 102 would normally be throttled, but because 101 failed, + // it should bypass throttling and try to write immediately! + when(mockOmResponse.getStatus()).thenReturn(OzoneManagerProtocolProtos.Status.OK); + ozoneFileCheckpointStrategy.save("102"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(3)); + + // Save 103 is throttled again (throttling restored) + ozoneFileCheckpointStrategy.save("103"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(3)); + } + } } From 450de80c8aa43b69b667294e6071a2f286e740d1 Mon Sep 17 00:00:00 2001 From: Colm Dougan Date: Thu, 2 Jul 2026 16:43:43 +0100 Subject: [PATCH 3/3] HDDS-15562. EventNotification: Address Copilot review feedback * OzoneFileCheckpointStrategy / OMEventListenerPluginManager: - Rename METDATA_KEY typo to METADATA_KEY. - Remove the unused IOmMetadataReader constructor parameter. - Simplify OzoneFileCheckpointStrategy initialization now that the unused IOmMetadataReader parameter is removed. * OMEventListenerLedgerPollerSeekPosition: - Add a fast-path bypass in set() to avoid redundant save writes when setting the seek position to its current value. * TestOMEventListenerLedgerPollerRecovery: - Add testSetToSameValueDoesNotTriggerSave to verify that the fast-path bypass works correctly. --- ...EventListenerLedgerPollerSeekPosition.java | 6 ++++++ ...stOMEventListenerLedgerPollerRecovery.java | 20 +++++++++++++++++++ .../OMEventListenerPluginManager.java | 5 ++--- .../OzoneFileCheckpointStrategy.java | 11 +++++----- .../TestOzoneFileCheckpointStrategy.java | 5 +---- 5 files changed, 34 insertions(+), 13 deletions(-) diff --git a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java index 8fe4107a31dc..0a7c2e55606a 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java +++ b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java @@ -17,6 +17,7 @@ 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; @@ -75,6 +76,11 @@ public boolean verifyCheckpointAccess() { public void set(String val) { LOG.debug("Setting seek position {}", val); + String current = seekPosition.get(); + if (Objects.equals(current, val)) { + checkpointVerified = true; + return; + } try { if (checkpointStrategy != null) { checkpointStrategy.save(val); diff --git a/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java index 177638afc9fb..701f9e947a4e 100644 --- a/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java +++ b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java @@ -164,4 +164,24 @@ public void testSaveFailureUnverifiesCheckpoint() throws Exception { 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()); + } } diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java index 6e5b6c992866..c85288ecf102 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java @@ -93,9 +93,8 @@ static List loadAll(OzoneManager ozoneManager, OzoneConfigurati } NotificationCheckpointStrategy checkpointStrategy = null; - if (ozoneManager != null && ozoneManager.getOmMetadataReader() != null) { - checkpointStrategy = new OzoneFileCheckpointStrategy(ozoneManager, - ozoneManager.getOmMetadataReader().get(), conf); + if (ozoneManager != null) { + checkpointStrategy = new OzoneFileCheckpointStrategy(ozoneManager, conf); } OMEventListenerPluginContext pluginContext = new OMEventListenerPluginContextImpl(ozoneManager, checkpointStrategy); diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java index 9360d64df784..2bbe57345270 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java @@ -24,7 +24,6 @@ import org.apache.commons.lang3.StringUtils; import org.apache.hadoop.hdds.conf.OzoneConfiguration; import org.apache.hadoop.hdds.protocol.proto.HddsProtos; -import org.apache.hadoop.ozone.om.IOmMetadataReader; import org.apache.hadoop.ozone.om.OzoneManager; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; @@ -69,7 +68,7 @@ public class OzoneFileCheckpointStrategy implements NotificationCheckpointStrate "ozone.om.plugin.kafka.checkpoint.save.interval"; public static final int OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL_DEFAULT = 100; - private static final String METDATA_KEY = "notification-checkpoint"; + private static final String METADATA_KEY = "notification-checkpoint"; private final AtomicLong callId = new AtomicLong(0); private final ClientId clientId = ClientId.randomId(); @@ -80,7 +79,7 @@ public class OzoneFileCheckpointStrategy implements NotificationCheckpointStrate private final String bucket; private final int saveInterval; - public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, final IOmMetadataReader omMetadataReader, + public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, OzoneConfiguration conf) { this.ozoneManager = ozoneManager; this.volume = conf.get(OZONE_OM_PLUGIN_CHECKPOINT_VOLUME, OZONE_OM_PLUGIN_CHECKPOINT_VOLUME_DEFAULT); @@ -99,7 +98,7 @@ public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, final IOmMetadataR public String load() throws IOException { OmBucketInfo bucketInfo = ozoneManager.getBucketInfo(volume, bucket); if (bucketInfo != null && bucketInfo.getMetadata() != null) { - return bucketInfo.getMetadata().get(METDATA_KEY); + return bucketInfo.getMetadata().get(METADATA_KEY); } return null; } @@ -128,7 +127,7 @@ private void saveImpl(String val) throws IOException { .setVolumeName(volume) .setBucketName(bucket) .addMetadata(HddsProtos.KeyValue.newBuilder() - .setKey(METDATA_KEY) + .setKey(METADATA_KEY) .setValue(val) .build()) .build(); @@ -145,7 +144,7 @@ private void saveImpl(String val) throws IOException { .build(); submitRequest(omRequest); - LOG.info("Persisted {} = {} directly as metadata on bucket /{}/{}", METDATA_KEY, val, volume, bucket); + LOG.info("Persisted {} = {} directly as metadata on bucket /{}/{}", METADATA_KEY, val, volume, bucket); } private UserInfo getUserInfo() { diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java index d83cc53b6c64..3563bcffa042 100644 --- a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java @@ -23,7 +23,6 @@ import com.google.protobuf.ServiceException; import java.io.IOException; -import org.apache.hadoop.ozone.om.IOmMetadataReader; import org.apache.hadoop.ozone.om.OzoneManager; import org.apache.hadoop.ozone.om.helpers.OmBucketInfo; import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; @@ -47,8 +46,6 @@ public class TestOzoneFileCheckpointStrategy { @Mock private OzoneManager mockOzoneManager; @Mock - private IOmMetadataReader mockOmMetadataReader; - @Mock private OzoneManagerProtocolProtos.OMResponse mockOmResponse; @Mock private OmBucketInfo mockBucketInfo; @@ -57,7 +54,7 @@ public class TestOzoneFileCheckpointStrategy { @BeforeEach public void setup() { - ozoneFileCheckpointStrategy = new OzoneFileCheckpointStrategy(mockOzoneManager, mockOmMetadataReader, + ozoneFileCheckpointStrategy = new OzoneFileCheckpointStrategy(mockOzoneManager, new org.apache.hadoop.hdds.conf.OzoneConfiguration()); }