HDDS-13513. Ozone Event Notification

[gardenia]

Please describe your PR in detail:

A rough draft implementation of this outline in the design doc: #8871 (comment)

NOTE: This PR is not intended to be ready for merging but is being shared as a way to get early feedback on the gist of the approach and help drive the discussion of the design (#8871) There are many TODOs, rough edges and things that need to be fleshed out.

There are 2 logical parts:

1. in OzoneManagerStateMachine when certain write requests complete successfully a summary of that request is written to a new rocksdb "ledger" table named CompletedRequestInfo

The current CompletedRequestInfo schema is minimal:

/**
* CompletedRequestInfo table entry
*/
message CompletedRequestInfo {
optional int64 trxLogIndex = 1;
required Type cmdType = 2; // Type of the command
optional string volumeName = 3;
optional string bucketName = 4;
optional string keyName = 5;
optional uint64 creationTime = 6;
optional CreateKeyOperationArgs createKeyArgs = 7;
optional RenameKeyOperationArgs renameKeyArgs = 8;
optional DeleteKeyOperationArgs deleteKeyArgs = 9;
optional CommitKeyOperationArgs commitKeyArgs = 10;
optional CreateDirectoryOperationArgs createDirectoryArgs = 11;
optional CreateFileOperationArgs createFileArgs = 12;
}

The use of "optional" for the arguments was based on feedback from the community call where @errose28 made the point that the previous sketch of a schema (using a freeform Map<String, String>) did not jibe well with schema management and this optional pattern had been used in other places to get around that.

2. there is then a concept of an "event listener" plugin which can consume the records in the CompletedRequestInfo table.

• plugins implement a new interface: OMEventListener
• a helper class OMEventListenerLedgerPoller is provided which plugin implementations can use to periodically poll for newly written CompletedRequestInfo records table and passes them them to a callback.
• OMEventListenerKafkaPublisher is a concrete implementation of OMEventListener which consumes the latest CompletedRequestInfo records, serializes them to appropriate S3 style event notification and sends them to a configured kafka broker.
• there is a draft implementation of a persistence strategy for the seek position (i.e. the latest CompletedRequestInfo consumed and processed by the plugin) called LocalFileCheckpointStrategy. This is an ultimately flawed approach in the case of leader changes but still meets the minimal criteria of "at least once" until we implement something better.
• additionally we also have an implementation of the checkpoint strategy which stores the file on the ozone filesystem OzoneFileCheckpointStrategy (rather than a local file)
• there is a crude strawman/draft implementation of a strategy to rotate out old records for the ledger based on number of rows. This needs fleshed out and made robust/efficient.
• plugins can be loaded/configured dynamically similarly to ranger plugins, e.g.:

ozone.om.plugin.destination.kafka=true
ozone.om.plugin.destination.kafka.classname=org.apache.hadoop.ozone.om.eventlistener.OMEventListenerKafkaPublisher
ozone.notify.kafka.topic=test123
ozone.notify.kafka.bootstrap.servers=kafka-3:29092,kafka-1:29092,kafka-2:29092

TODO:

• move the plugin implementations out of ozone-manager into some separate mvn package
• implement a better persistence strategy for the seek position (e.g. write to a file on the ozone filesystem instead of a file local to the OM)
• scrub of terminology / renaming. (original terminology OperationInfo now renamed to CompletedRequestInfo / OmCompletedRequestInfo)
• flesh out the event mappings - the existing mappings of operations -> events are not an authoritative take
• flesh out/redo the strawman implementation of the strategy to rotate out old records from the ledger table (CompletedRequestInfoCleanupService)
• move the pluginManager creation/lifecycle out of KeyManagerImpl to some better place
• need to work out how to fit ACLs into the ledger schema
• metrics
• more unit tests

What is the link to the Apache JIRA

http://issues.apache.org/jira/browse/HDDS-5984

How was this patch tested?

unit tests, manual tests (docker compose)

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[github-actions]

Thank you for your contribution. This PR is being closed due to inactivity. If needed, feel free to reopen it.

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[smengcl]

Code review

This is an early-stage design/feedback PR (author's own description: "not intended to be ready for merging"). Reviewing in that spirit. Note that PR #8871 by a different author also implements HDDS-13513 -- the community will need to reconcile which design to pursue before either can land.

Found 4 issues:

1. equals() compares the wrong fields with == instead of .equals(), making the method return incorrect results for most inputs.

volumeName == that.bucketName && // wrong: reference equality and wrong field pairing

ozone/hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmCompletedRequestInfo.java

Lines 373 to 380 in 047e955

	bucketName.equals(that.bucketName) &&
	keyName.equals(that.keyName) &&
	volumeName == that.bucketName &&
	opArgs.equals(that.opArgs);
	}
	@Override
	publicinthashCode() {

2. CompletedRequestInfoCleanupService calls deleteRange(firstKey, lastKey) but RocksDB deleteRange is end-exclusive ([first, last)), so the last record intended for deletion is always skipped. The table grows by 1 row every cleanup cycle. The single-key path (delete(firstKey)) is correct; the range path needs deleteRange(firstKey, keyAfterLast) or an explicit delete of lastKey after the range call.

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

Lines 186 to 192 in 047e955

	if (Objects.equals(firstKeyToDelete, lastKeyToDelete)) {
	metadataManager.getCompletedRequestInfoTable().delete(firstKeyToDelete);
	} else {
	metadataManager.getCompletedRequestInfoTable().deleteRange(firstKeyToDelete, lastKeyToDelete);
	}
	} catch (IOExceptione) {

3. Several new files are missing the Apache license header. CLAUDE.md requires it on all new files, and rat.sh will fail. Affected files include NotificationCheckpointStrategy.java, OMEventListener.java, OMEventListenerPluginContext.java, LocalFileCheckpointStrategy.java, OMEventListenerPluginContextImpl.java, OzoneFileCheckpointStrategy.java, OMEventListenerNotificationStrategy.java, S3EventNotificationBuilder.java, and S3EventNotificationStrategy.java.

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

Lines 1 to 5 in 047e955

	packageorg.apache.hadoop.ozone.om.eventlistener;
	importjava.io.IOException;
	/**

4. S3EventNotification.java and DateTimeJsonSerializer.java carry Amazon copyright headers (Copyright 2014-2025 Amazon Technologies, Inc. / Copyright (c) 2016. Amazon.com, Inc.) copied from the AWS SDK 1.x. The ASF RAT check will reject these. The PR comment already notes "We may not need to fork this class" -- if the class is needed, it should be rewritten or the provenance documented under ASF IP policy.

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

Lines 1 to 12 in 047e955

	/*
	* Copyright 2014-2025 Amazon Technologies, Inc.
	*
	* Licensed 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://aws.amazon.com/apache2.0
	*
	* This file 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

🤖 Generated with Claude Code

_{- If this code review was useful, please react with 👍. Otherwise, react with 👎.}

[github-actions]

This PR has been marked as stale due to 21 days of inactivity. Please comment or remove the stale label to keep it open. Otherwise, it will be automatically closed in 7 days.

[github-actions]

Thank you for your contribution. This PR is being closed due to inactivity. Please contact a maintainer if you would like to reopen it.