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/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 bccbda1e2a7e..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; @@ -24,21 +25,27 @@ /** * 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; + 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; } @@ -46,12 +53,47 @@ 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 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..701f9e947a4e --- /dev/null +++ b/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 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()); + } +} 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..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 @@ -92,7 +92,11 @@ static List 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 { 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..2bbe57345270 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java @@ -0,0 +1,183 @@ +/* + * 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.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.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; +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 METADATA_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, + 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 { + OmBucketInfo bucketInfo = ozoneManager.getBucketInfo(volume, bucket); + if (bucketInfo != null && bucketInfo.getMetadata() != null) { + return bucketInfo.getMetadata().get(METADATA_KEY); + } + 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) { + 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) throws IOException { + BucketArgs bucketArgs = BucketArgs.newBuilder() + .setVolumeName(volume) + .setBucketName(bucket) + .addMetadata(HddsProtos.KeyValue.newBuilder() + .setKey(METADATA_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 /{}/{}", METADATA_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) throws IOException { + try { + 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.", e); + throw new IOException(e); + } + } +} 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..3563bcffa042 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java @@ -0,0 +1,158 @@ +/* + * 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.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 OzoneManagerProtocolProtos.OMResponse mockOmResponse; + @Mock + private OmBucketInfo mockBucketInfo; + + private OzoneFileCheckpointStrategy ozoneFileCheckpointStrategy; + + @BeforeEach + public void setup() { + ozoneFileCheckpointStrategy = new OzoneFileCheckpointStrategy(mockOzoneManager, + 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); + when(mockOmResponse.getStatus()).thenReturn(OzoneManagerProtocolProtos.Status.OK); + + // 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.assertThrows(IOException.class, () -> 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()); + } + + @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)); + } + } +}