Copilot commented on code in PR #10689: URL: https://github.com/apache/ozone/pull/10689#discussion_r3572490195
########## 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; + Review Comment: validateAndUpdateCache() will happily process a malformed OMRequest that omits setEventNotificationCheckpointRequest (proto2 optional field) or provides an empty checkpointKey, which results in writing to META_TABLE under just the prefix ("#EVENTNOTIFICATIONCHECKPOINT#") and can unintentionally clobber/poison the metadata keyspace. Please reject invalid requests before updating the cache/batch. ########## hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java: ########## @@ -33,6 +33,8 @@ public interface OMEventListenerPluginContext { // them to some predefined value for safety? e.g. 10K List<OmCompletedRequestInfo> listCompletedRequestInfo(Long startKey, int maxResults) throws IOException; + NotificationCheckpointStrategy getNotificationCheckpointStrategy(); + Review Comment: getNotificationCheckpointStrategy() can return null (eg OMEventListenerPluginManager passes null when ozoneManager is null), but the interface contract doesn’t document this. This makes it easy for plugin implementations to accidentally dereference it and crash. Please either (a) document/annotate it as @Nullable, or (b) return a no-op strategy implementation to avoid nulls. ########## 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); Review Comment: saveImpl() logs every persisted checkpoint at INFO. Even with throttling (default 100), this can still be very noisy on busy clusters and inflate OM logs. Consider lowering this to DEBUG (or rate limiting) since it’s routine progress output. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
