diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java index 3bd8388f9500..a8dc240dda00 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java @@ -133,6 +133,9 @@ public final class OzoneConsts { public static final String RANGER_OZONE_SERVICE_VERSION_KEY = "#RANGEROZONESERVICEVERSION"; + public static final String EVENT_NOTIFICATION_CHECKPOINT_PREFIX = + "#EVENTNOTIFICATIONCHECKPOINT#"; + public static final String MULTIPART_FORM_DATA_BOUNDARY = "---XXX"; // Block ID prefixes used in datanode containers. diff --git a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java index af89cd2ddb07..2554f6327215 100644 --- a/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OmUtils.java @@ -307,6 +307,7 @@ public static boolean isReadOnly(OMRequest omRequest) { case TenantAssignAdmin: case TenantRevokeAdmin: case SetRangerServiceVersion: + case SetEventNotificationCheckpoint: case CreateSnapshot: case DeleteSnapshot: case RenameSnapshot: @@ -422,6 +423,7 @@ public static boolean shouldSendToFollower(OMRequest omRequest) { case TenantAssignAdmin: case TenantRevokeAdmin: case SetRangerServiceVersion: + case SetEventNotificationCheckpoint: case CreateSnapshot: case DeleteSnapshot: case RenameSnapshot: 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..7b8c984274fd --- /dev/null +++ b/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java @@ -0,0 +1,33 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.om.eventlistener; + +import java.io.IOException; + +/** + * Interface for implementations which load/save the current checkpoint + * transaction log index used by an event poller. + */ +public interface NotificationCheckpointStrategy { + + String load() throws IOException; + + void save(String val) throws IOException; + + void reset() throws IOException; +} 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..777ebc169179 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 @@ -33,6 +33,8 @@ public interface OMEventListenerPluginContext { // them to some predefined value for safety? e.g. 10K List listCompletedRequestInfo(Long startKey, int maxResults) throws IOException; + NotificationCheckpointStrategy getNotificationCheckpointStrategy(); + // XXX: this probably doesn't belong here String getThreadNamePrefix(); } diff --git a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto index 12708817c31d..d5b1597ea479 100644 --- a/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto +++ b/hadoop-ozone/interface-client/src/main/proto/OmClientProtocol.proto @@ -156,6 +156,7 @@ enum Type { PutObjectTagging = 140; GetObjectTagging = 141; DeleteObjectTagging = 142; + SetEventNotificationCheckpoint = 144; } enum SafeMode { @@ -304,6 +305,7 @@ message OMRequest { optional PutObjectTaggingRequest putObjectTaggingRequest = 141; optional DeleteObjectTaggingRequest deleteObjectTaggingRequest = 142; repeated SetSnapshotPropertyRequest SetSnapshotPropertyRequests = 143; + optional SetEventNotificationCheckpointRequest setEventNotificationCheckpointRequest = 144; } message OMResponse { @@ -437,6 +439,7 @@ message OMResponse { optional GetObjectTaggingResponse getObjectTaggingResponse = 140; optional PutObjectTaggingResponse putObjectTaggingResponse = 141; optional DeleteObjectTaggingResponse deleteObjectTaggingResponse = 142; + optional SetEventNotificationCheckpointResponse setEventNotificationCheckpointResponse = 144; } enum Status { @@ -2003,6 +2006,11 @@ message SetRangerServiceVersionRequest { required uint64 rangerServiceVersion = 1; } +message SetEventNotificationCheckpointRequest { + required string checkpointKey = 1; + required string checkpointValue = 2; +} + message CreateSnapshotRequest { optional string volumeName = 1; optional string bucketName = 2; @@ -2156,6 +2164,9 @@ message CreateTenantResponse { message SetRangerServiceVersionResponse { } +message SetEventNotificationCheckpointResponse { +} + message CreateSnapshotResponse { optional SnapshotInfo snapshotInfo = 1; } 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..9f1a14a219b4 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 NotificationCheckpointStrategy checkpointStrategy; private final AtomicReference seekPosition; - public OMEventListenerLedgerPollerSeekPosition() { - this.seekPosition = new AtomicReference(initSeekPosition()); + public OMEventListenerLedgerPollerSeekPosition(NotificationCheckpointStrategy checkpointStrategy) { + this.checkpointStrategy = checkpointStrategy; + this.seekPosition = new AtomicReference<>(initSeekPosition()); } - // TODO: load this from persistent storage public String initSeekPosition() { + try { + if (checkpointStrategy != null) { + return checkpointStrategy.load(); + } + } catch (Exception ex) { + LOG.warn("Failed to load initial seek position from checkpoint strategy. " + + "Fallback to starting from the beginning.", ex); + } return null; } @@ -48,10 +55,36 @@ public String get() { public void set(String val) { LOG.debug("Setting seek position {}", val); - // NOTE: this in-memory view of the seek position needs to be kept - // up to date because the OMEventListenerLedgerPoller has a - // reference to it - seekPosition.set(val); + String current = seekPosition.get(); + if (Objects.equals(current, val)) { + return; + } + try { + if (checkpointStrategy != null) { + checkpointStrategy.save(val); + } + } catch (Exception ex) { + // Save failure does NOT block subsequent runs or in-memory progress! + LOG.warn("Failed to save seek position checkpoint {} to database. " + + "Progress will continue in-memory but is not durably saved.", val, ex); + } finally { + // ALWAYS advance the in-memory seek position, even if saving fails, + // so that the background task is not blocked from making progress. + seekPosition.set(val); + } + } + + public void reset() { + LOG.debug("Resetting seek position"); + try { + if (checkpointStrategy != null) { + checkpointStrategy.reset(); + } + } catch (Exception ex) { + LOG.warn("Failed to reset seek position checkpoint on database strategy.", ex); + } finally { + seekPosition.set(null); + } } @Override 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..79b40090fcea --- /dev/null +++ b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java @@ -0,0 +1,121 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hadoop.ozone.om.eventlistener; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.utils.BackgroundTask; +import org.apache.hadoop.hdds.utils.BackgroundTaskQueue; +import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo; +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +/** + * Tests for OMEventListenerLedgerPoller recovery and checkpoint non-blocking behavior. + */ +@ExtendWith(MockitoExtension.class) +public class TestOMEventListenerLedgerPollerRecovery { + + @Mock + private OMEventListenerPluginContext pluginContext; + + @Mock + private NotificationCheckpointStrategy checkpointStrategy; + + @Mock + private Consumer callback; + + @Test + public void testSaveFailureIsCompletelyNonBlocking() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + + when(pluginContext.isLeaderReady()).thenReturn(true); + when(pluginContext.getThreadNamePrefix()).thenReturn("test-poller-"); + + // Mock load to return 10 + doReturn("10").when(checkpointStrategy).load(); + + OMEventListenerLedgerPollerSeekPosition seekPosition = + new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy); + + // Initial position in memory should be 10 + Assertions.assertEquals("10", seekPosition.get()); + + OMEventListenerLedgerPoller poller = new OMEventListenerLedgerPoller( + 1000, TimeUnit.MILLISECONDS, 1, 1000, + pluginContext, conf, seekPosition, callback); + + BackgroundTaskQueue queue = poller.getTasks(); + BackgroundTask task = queue.poll(); + + // 1. First poller run: succeeds + task.call(); + verify(pluginContext).listCompletedRequestInfo(any(), anyInt()); + + // 2. Now simulate checkpoint strategy becoming unhealthy (save throws IOException) + doThrow(new IOException("Save failed")).when(checkpointStrategy).save("11"); + seekPosition.set("11"); + + // In-memory view should STILL be advanced to "11" so we don't block progress + Assertions.assertEquals("11", seekPosition.get()); + } + + @Test + public void testLoadFailureOnStartupFallbackToNull() throws Exception { + OzoneConfiguration conf = new OzoneConfiguration(); + + // Mock load to fail during initialization / startup + doThrow(new IOException("Database down")).when(checkpointStrategy).load(); + + OMEventListenerLedgerPollerSeekPosition seekPosition = + new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy); + + // Initial load failure means we fallback to starting from the beginning (null) + Assertions.assertNull(seekPosition.get()); + } + + @Test + public void testSetToSameValueDoesNotTriggerSave() throws Exception { + doReturn("10").when(checkpointStrategy).load(); + + OMEventListenerLedgerPollerSeekPosition seekPosition = + new OMEventListenerLedgerPollerSeekPosition(checkpointStrategy); + + // Initial position in memory should be "10" + Assertions.assertEquals("10", seekPosition.get()); + + // Setting to the same value "10" should skip calling save on checkpointStrategy + seekPosition.set("10"); + + // Verify checkpointStrategy.save("10") was never called + verify(checkpointStrategy, never()).save("10"); + } +} 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..eb1856277ab0 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 @@ -45,6 +48,11 @@ public List listCompletedRequestInfo(Long startKey, int return ozoneManager.getMetadataManager().listCompletedRequestInfo(startKey, maxResults); } + @Override + public NotificationCheckpointStrategy getNotificationCheckpointStrategy() { + return checkpointStrategy; + } + // TODO: it feels like this doesn't belong here @Override public String getThreadNamePrefix() { 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..e2d6f0fad2b0 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 OzoneDbCheckpointStrategy(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/OzoneDbCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneDbCheckpointStrategy.java new file mode 100644 index 000000000000..02149ac08163 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneDbCheckpointStrategy.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.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerRatisUtils; +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.SetEventNotificationCheckpointRequest; +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 last known transaction progress directly in the OM DB's metaTable. + * + * This allows lightweight, HA-consistent, and isolated checkpointing + * without filesystem volumes, buckets, or client-contended locks. + */ +public class OzoneDbCheckpointStrategy implements NotificationCheckpointStrategy { + + public static final Logger LOG = LoggerFactory.getLogger(OzoneDbCheckpointStrategy.class); + + public static final String OZONE_OM_PLUGIN_CHECKPOINT_NAME = "ozone.om.plugin.kafka.checkpoint.name"; + public static final String OZONE_OM_PLUGIN_CHECKPOINT_NAME_DEFAULT = "kafka-completed-ops"; + + 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 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 checkpointName; + private final String dbKey; + private final int saveInterval; + + public OzoneDbCheckpointStrategy(OzoneManager ozoneManager, + OzoneConfiguration conf) { + this.ozoneManager = ozoneManager; + this.checkpointName = conf.get(OZONE_OM_PLUGIN_CHECKPOINT_NAME, OZONE_OM_PLUGIN_CHECKPOINT_NAME_DEFAULT); + this.dbKey = OzoneConsts.EVENT_NOTIFICATION_CHECKPOINT_PREFIX + checkpointName; + + 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 { + return ozoneManager.getMetadataManager().getMetaTable().get(dbKey); + } + + @Override + public void save(String val) throws IOException { + if (StringUtils.isBlank(val)) { + return; + } + long currentSaveCount = saveCount.get(); + // Throttle database commits: persist checkpoint based on configured interval to avoid write storms + if (currentSaveCount == 0 || currentSaveCount % saveInterval == 0) { + try { + saveImpl(val); + // Successful save, so increment count + saveCount.incrementAndGet(); + } catch (IOException e) { + // If save fails, do not increment saveCount so we retry next time + throw e; + } + } else { + // If we are throttled, we still increment saveCount to keep track of updates + saveCount.incrementAndGet(); + } + } + + @Override + public void reset() throws IOException { + try { + saveImpl(""); + // Reset count so next normal save starts fresh + saveCount.set(0); + } catch (IOException e) { + throw e; + } + } + + private void saveImpl(String val) throws IOException { + SetEventNotificationCheckpointRequest setCheckpointRequest = SetEventNotificationCheckpointRequest.newBuilder() + .setCheckpointKey(checkpointName) + .setCheckpointValue(val) + .build(); + + OMRequest omRequest = OMRequest.newBuilder() + .setCmdType(Type.SetEventNotificationCheckpoint) + .setClientId(clientId.toString()) + .setSetEventNotificationCheckpointRequest(setCheckpointRequest) + .setUserInfo(getUserInfo()) + .build(); + + submitRequest(omRequest); + LOG.info("Persisted {} = {} directly as metadata inside metaTable under key {}", + checkpointName, val, dbKey); + } + + 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() : "")); + } + if (response == null) { + throw new IOException("Failed to persist checkpoint: empty response"); + } + return response; + } catch (ServiceException e) { + LOG.error("Set event notification checkpoint " + omRequest.getCmdType() + " request failed.", e); + throw new IOException(e); + } + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java index 1778de3520d9..491c40ed5171 100644 --- a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/ratis/utils/OzoneManagerRatisUtils.java @@ -50,6 +50,7 @@ import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketAddAclRequest; import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketRemoveAclRequest; import org.apache.hadoop.ozone.om.request.bucket.acl.OMBucketSetAclRequest; +import org.apache.hadoop.ozone.om.request.eventlistener.OMSetEventNotificationCheckpointRequest; import org.apache.hadoop.ozone.om.request.file.OMRecoverLeaseRequest; import org.apache.hadoop.ozone.om.request.key.OMDirectoriesPurgeRequestWithFSO; import org.apache.hadoop.ozone.om.request.key.OMKeyPurgeRequest; @@ -217,6 +218,8 @@ public static OMClientRequest createClientRequest(OMRequest omRequest, return new OMTenantRevokeAdminRequest(omRequest); case SetRangerServiceVersion: return new OMSetRangerServiceVersionRequest(omRequest); + case SetEventNotificationCheckpoint: + return new OMSetEventNotificationCheckpointRequest(omRequest); case CreateSnapshot: return new OMSnapshotCreateRequest(omRequest); case DeleteSnapshot: diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/OMSetEventNotificationCheckpointRequest.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/OMSetEventNotificationCheckpointRequest.java new file mode 100644 index 000000000000..7a52798f59f1 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/OMSetEventNotificationCheckpointRequest.java @@ -0,0 +1,77 @@ +/* + * 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.request.eventlistener; + +import org.apache.hadoop.hdds.utils.db.cache.CacheKey; +import org.apache.hadoop.hdds.utils.db.cache.CacheValue; +import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.execution.flowcontrol.ExecutionContext; +import org.apache.hadoop.ozone.om.request.OMClientRequest; +import org.apache.hadoop.ozone.om.request.util.OmResponseUtil; +import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.om.response.eventlistener.OMSetEventNotificationCheckpointResponse; +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.SetEventNotificationCheckpointRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetEventNotificationCheckpointResponse; + +/** + * Handles OMSetEventNotificationCheckpointRequest. + * + * This is an Ozone Manager internal request used to persist event notification checkpoints + * inside the metaTable. + */ +public class OMSetEventNotificationCheckpointRequest extends OMClientRequest { + + public OMSetEventNotificationCheckpointRequest(OMRequest omRequest) { + super(omRequest); + } + + @Override + public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager, ExecutionContext context) { + + OMClientResponse omClientResponse; + final OMResponse.Builder omResponse = + OmResponseUtil.getOMResponseBuilder(getOmRequest()); + OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager(); + final SetEventNotificationCheckpointRequest request = + getOmRequest().getSetEventNotificationCheckpointRequest(); + + final String checkpointKey = request.getCheckpointKey(); + final String checkpointValue = request.getCheckpointValue(); + + // Store in metaTable using CacheKey with our specific checkpoint prefix + final String dbKey = OzoneConsts.EVENT_NOTIFICATION_CHECKPOINT_PREFIX + checkpointKey; + + omMetadataManager.getMetaTable().addCacheEntry( + new CacheKey<>(dbKey), + CacheValue.get(context.getIndex(), checkpointValue)); + + omResponse.setSetEventNotificationCheckpointResponse( + SetEventNotificationCheckpointResponse.newBuilder().build()); + + omClientResponse = new OMSetEventNotificationCheckpointResponse( + omResponse.build(), + dbKey, + checkpointValue); + + return omClientResponse; + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/package-info.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/package-info.java new file mode 100644 index 000000000000..8f95ba84f743 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/eventlistener/package-info.java @@ -0,0 +1,21 @@ +/* + * 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. + */ + +/** + * This package contains classes for event listener requests. + */ +package org.apache.hadoop.ozone.om.request.eventlistener; diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/OMSetEventNotificationCheckpointResponse.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/OMSetEventNotificationCheckpointResponse.java new file mode 100644 index 000000000000..b1273330f868 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/OMSetEventNotificationCheckpointResponse.java @@ -0,0 +1,61 @@ +/* + * 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.response.eventlistener; + +import static org.apache.hadoop.ozone.om.codec.OMDBDefinition.META_TABLE; + +import jakarta.annotation.Nonnull; +import java.io.IOException; +import org.apache.hadoop.hdds.utils.db.BatchOperation; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.apache.hadoop.ozone.om.response.CleanupTableInfo; +import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMResponse; + +/** + * Response for OMSetEventNotificationCheckpointRequest. + */ +@CleanupTableInfo(cleanupTables = {META_TABLE}) +public class OMSetEventNotificationCheckpointResponse extends OMClientResponse { + private String checkpointDbKey; + private String checkpointValue; + + public OMSetEventNotificationCheckpointResponse(@Nonnull OMResponse omResponse, + @Nonnull String dbKey, + @Nonnull String val) { + super(omResponse); + this.checkpointDbKey = dbKey; + this.checkpointValue = val; + } + + /** + * For when the request is not successful. + */ + public OMSetEventNotificationCheckpointResponse(@Nonnull OMResponse omResponse) { + super(omResponse); + checkStatusNotOK(); + } + + @Override + public void addToDBBatch(OMMetadataManager omMetadataManager, + BatchOperation batchOperation) throws IOException { + + omMetadataManager.getMetaTable().putWithBatch( + batchOperation, checkpointDbKey, checkpointValue); + } +} diff --git a/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/package-info.java b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/package-info.java new file mode 100644 index 000000000000..2df434d33183 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/response/eventlistener/package-info.java @@ -0,0 +1,21 @@ +/* + * 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. + */ + +/** + * This package contains classes for event listener responses. + */ +package org.apache.hadoop.ozone.om.response.eventlistener; diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneDbCheckpointStrategy.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneDbCheckpointStrategy.java new file mode 100644 index 000000000000..bf670686fbc7 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneDbCheckpointStrategy.java @@ -0,0 +1,161 @@ +/* + * 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.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.utils.db.Table; +import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.OMMetadataManager; +import org.apache.hadoop.ozone.om.OzoneManager; +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 OzoneDbCheckpointStrategy}. + */ +@ExtendWith(MockitoExtension.class) +public class TestOzoneDbCheckpointStrategy { + + @Mock + private OzoneManager mockOzoneManager; + @Mock + private OMMetadataManager mockMetadataManager; + @Mock + @SuppressWarnings("rawtypes") + private Table mockMetaTable; + @Mock + private OzoneManagerProtocolProtos.OMResponse mockOmResponse; + + private OzoneDbCheckpointStrategy ozoneDbCheckpointStrategy; + + @BeforeEach + public void setup() throws Exception { + ozoneDbCheckpointStrategy = new OzoneDbCheckpointStrategy(mockOzoneManager, + new OzoneConfiguration()); + } + + @Test + @SuppressWarnings("unchecked") + public void testLoadStrategy() throws IOException { + when(mockOzoneManager.getMetadataManager()).thenReturn(mockMetadataManager); + when(mockMetadataManager.getMetaTable()).thenReturn(mockMetaTable); + String dbKey = OzoneConsts.EVENT_NOTIFICATION_CHECKPOINT_PREFIX + "kafka-completed-ops"; + when(mockMetaTable.get(dbKey)).thenReturn("00000000000000000017"); + Assertions.assertEquals("00000000000000000017", ozoneDbCheckpointStrategy.load()); + } + + @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 + ozoneDbCheckpointStrategy.save("00000000000000000001"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + // Iterations 2 to 100 are throttled + for (int i = 2; i <= 100; i++) { + ozoneDbCheckpointStrategy.save(String.valueOf(i)); + } + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + + // Iteration 101 should write immediately! + ozoneDbCheckpointStrategy.save("00000000000000000101"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(2)); + } + } + + @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); + ozoneDbCheckpointStrategy.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++) { + ozoneDbCheckpointStrategy.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, () -> ozoneDbCheckpointStrategy.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); + ozoneDbCheckpointStrategy.save("102"); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(3)); + } + } + + @Test + public void testExplicitReset() 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); + + // Calling reset() writes empty string immediately, even if throttled count is non-zero + ozoneDbCheckpointStrategy.reset(); + utils.verify(() -> OzoneManagerRatisUtils.submitRequest(any(OzoneManager.class), + any(OzoneManagerProtocolProtos.OMRequest.class), + any(ClientId.class), any(Long.class)), Mockito.times(1)); + } + } +} diff --git a/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/eventlistener/TestOMSetEventNotificationCheckpointRequest.java b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/eventlistener/TestOMSetEventNotificationCheckpointRequest.java new file mode 100644 index 000000000000..f496f985af46 --- /dev/null +++ b/hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/request/eventlistener/TestOMSetEventNotificationCheckpointRequest.java @@ -0,0 +1,112 @@ +/* + * 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.request.eventlistener; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; +import static org.mockito.Mockito.framework; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.io.IOException; +import java.nio.file.Path; +import java.util.UUID; +import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.ozone.OzoneConsts; +import org.apache.hadoop.ozone.om.OMConfigKeys; +import org.apache.hadoop.ozone.om.OMPerformanceMetrics; +import org.apache.hadoop.ozone.om.OmMetadataManagerImpl; +import org.apache.hadoop.ozone.om.OzoneManager; +import org.apache.hadoop.ozone.om.response.OMClientResponse; +import org.apache.hadoop.ozone.om.response.eventlistener.OMSetEventNotificationCheckpointResponse; +import org.apache.hadoop.ozone.om.upgrade.OMLayoutVersionManager; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OMRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.SetEventNotificationCheckpointRequest; +import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Tests OMSetEventNotificationCheckpointRequest. + */ +public class TestOMSetEventNotificationCheckpointRequest { + + @TempDir + private Path folder; + + private OzoneManager ozoneManager; + + @BeforeEach + public void setUp() throws Exception { + ozoneManager = mock(OzoneManager.class); + when(ozoneManager.getVersionManager()) + .thenReturn(new OMLayoutVersionManager(1)); + + final OzoneConfiguration conf = new OzoneConfiguration(); + conf.set(OMConfigKeys.OZONE_OM_DB_DIRS, + folder.toAbsolutePath().toString()); + OmMetadataManagerImpl omMetadataManager = new OmMetadataManagerImpl(conf, + ozoneManager); + when(ozoneManager.getMetadataManager()) + .thenReturn(omMetadataManager); + OMPerformanceMetrics omPerformanceMetrics = mock(OMPerformanceMetrics.class); + when(ozoneManager.getPerfMetrics()).thenReturn(omPerformanceMetrics); + } + + @AfterEach + public void tearDown() throws Exception { + framework().clearInlineMocks(); + } + + private OMRequest createSetCheckpointRequest(String key, String value) { + return OMRequest.newBuilder() + .setClientId(UUID.randomUUID().toString()) + .setCmdType(Type.SetEventNotificationCheckpoint) + .setSetEventNotificationCheckpointRequest( + SetEventNotificationCheckpointRequest.newBuilder() + .setCheckpointKey(key) + .setCheckpointValue(value) + .build()) + .build(); + } + + @Test + public void testRequest() throws IOException { + long txLogIndex = 1; + + // Run preExecute + OMSetEventNotificationCheckpointRequest request = + new OMSetEventNotificationCheckpointRequest( + new OMSetEventNotificationCheckpointRequest( + createSetCheckpointRequest("kafka-completed-ops", "100")).preExecute(ozoneManager)); + + // Run validateAndUpdateCache + OMClientResponse clientResponse = request.validateAndUpdateCache( + ozoneManager, txLogIndex); + + // Check response type + assertInstanceOf(OMSetEventNotificationCheckpointResponse.class, clientResponse); + + // Verify response caches correctly and writes correctly to DB + String dbKey = OzoneConsts.EVENT_NOTIFICATION_CHECKPOINT_PREFIX + "kafka-completed-ops"; + String valInDb = ozoneManager.getMetadataManager().getMetaTable().get(dbKey); + assertEquals("100", valInDb); + } +}