This is an automated email from the ASF dual-hosted git repository.
ChenSammi pushed a commit to branch HDDS-13513_Event_Notification_FeatureBranch
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to
refs/heads/HDDS-13513_Event_Notification_FeatureBranch by this push:
new da6a6e0dc97 HDDS-14009. EventNotification: Generate events according
to S3 schema/strategy (#10318)
da6a6e0dc97 is described below
commit da6a6e0dc97537eee6db79d9a11c1857e882a606
Author: gardenia <[email protected]>
AuthorDate: Thu Jun 11 10:23:03 2026 +0100
HDDS-14009. EventNotification: Generate events according to S3
schema/strategy (#10318)
---
hadoop-ozone/ozone-manager-plugins/pom.xml | 12 +
.../OMEventListenerKafkaPublisher.java | 62 ++-
.../OMEventListenerNotificationStrategy.java | 30 ++
.../eventlistener/s3/DateTimeJsonSerializer.java | 36 ++
.../om/eventlistener/s3/S3EventNotification.java | 560 +++++++++++++++++++++
.../s3/S3EventNotificationBuilder.java | 128 +++++
.../s3/S3EventNotificationStrategy.java | 190 +++++++
.../ozone/om/eventlistener/s3/package-info.java | 22 +
.../TestOMEventListenerKafkaPublisher.java | 213 ++++++--
9 files changed, 1203 insertions(+), 50 deletions(-)
diff --git a/hadoop-ozone/ozone-manager-plugins/pom.xml
b/hadoop-ozone/ozone-manager-plugins/pom.xml
index 95abfcf4ef6..4fe596bb979 100644
--- a/hadoop-ozone/ozone-manager-plugins/pom.xml
+++ b/hadoop-ozone/ozone-manager-plugins/pom.xml
@@ -29,6 +29,18 @@
</properties>
<dependencies>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-annotations</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ </dependency>
<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
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 53724129b0d..f42524ba98a 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
@@ -18,12 +18,16 @@
package org.apache.hadoop.ozone.om.eventlistener;
import java.io.IOException;
+import java.time.Duration;
import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.OzoneIllegalArgumentException;
+import org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotificationStrategy;
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
@@ -42,10 +46,16 @@ public class OMEventListenerKafkaPublisher implements
OMEventListener {
public static final Logger LOG =
LoggerFactory.getLogger(OMEventListenerKafkaPublisher.class);
private static final String KAFKA_CONFIG_PREFIX = "ozone.om.plugin.kafka.";
+ private static final String NOTIFICATION_STRATEGY_CONFIG =
KAFKA_CONFIG_PREFIX + "notification.strategy";
+ private static final Class<? extends OMEventListenerNotificationStrategy>
+ DEFAULT_NOTIFICATION_STRATEGY = S3EventNotificationStrategy.class;
+ private static final String KAFKA_SERVICE_INTERVAL_CONFIG =
KAFKA_CONFIG_PREFIX + "service.interval";
+ private static final String KAFKA_SERVICE_TIMEOUT_CONFIG =
KAFKA_CONFIG_PREFIX + "service.timeout";
private static final int COMPLETED_REQUEST_CONSUMER_CORE_POOL_SIZE = 1;
private OMEventListenerLedgerPoller ledgerPoller;
private KafkaClientWrapper kafkaClient;
+ private OMEventListenerNotificationStrategy notificationStrategy;
private OMEventListenerLedgerPollerSeekPosition seekPosition;
@Override
@@ -56,10 +66,24 @@ public void initialize(OzoneConfiguration conf,
OMEventListenerPluginContext plu
this.kafkaClient = new KafkaClientWrapper(kafkaProps);
- // TODO: these constants should be read from config
- long kafkaServiceInterval = 2 * 1000;
- long kafkaServiceTimeout = 300 * 1000;
+ long kafkaServiceInterval = conf.getTimeDuration(
+ KAFKA_SERVICE_INTERVAL_CONFIG, "2s", TimeUnit.MILLISECONDS);
+ long kafkaServiceTimeout = conf.getTimeDuration(
+ KAFKA_SERVICE_TIMEOUT_CONFIG, "5m", TimeUnit.MILLISECONDS);
+ Class<? extends OMEventListenerNotificationStrategy> strategyClass =
conf.getClass(
+ NOTIFICATION_STRATEGY_CONFIG,
+ DEFAULT_NOTIFICATION_STRATEGY,
+ OMEventListenerNotificationStrategy.class);
+ try {
+ this.notificationStrategy =
strategyClass.getDeclaredConstructor().newInstance();
+ } catch (Exception ex) {
+ LOG.error("Failed to instantiate notification strategy: {}",
strategyClass, ex);
+ OzoneIllegalArgumentException exception = new
OzoneIllegalArgumentException(
+ "Failed to instantiate notification strategy: " + strategyClass);
+ exception.initCause(ex);
+ throw exception;
+ }
this.seekPosition = new OMEventListenerLedgerPollerSeekPosition();
LOG.info("Creating OMEventListenerLedgerPoller with serviceInterval={}," +
@@ -102,24 +126,23 @@ public void stop() {
public void handleCompletedRequest(OmCompletedRequestInfo
completedRequestInfo) {
LOG.debug("Processing {}", completedRequestInfo);
- // stub event until we implement a strategy to convert the events to
- // a user facing schema (e.g. S3)
- String event = String.format("{\"key\":\"%s/%s/%s\", \"type\":\"%s\"}",
- completedRequestInfo.getVolumeName(),
- completedRequestInfo.getBucketName(),
- completedRequestInfo.getKeyName(),
- String.valueOf(completedRequestInfo.getCmdType()));
+ List<String> eventsToSend =
notificationStrategy.determineEventsForOperation(completedRequestInfo);
- LOG.debug("Sending {}", event);
-
- try {
- kafkaClient.send(event);
- } catch (IOException ex) {
- LOG.error("Failure to send event {}", event, ex);
- return;
+ // loop over events and send them to our kafka sink
+ for (String event : eventsToSend) {
+ if (event == null) {
+ LOG.warn("Skipping null event for transaction {}",
completedRequestInfo.getTrxLogIndex());
+ continue;
+ }
+ try {
+ kafkaClient.send(event);
+ } catch (IOException ex) {
+ LOG.error("Failure to send event {}", event, ex);
+ return;
+ }
}
- // we can update the seek position
+ // no errors so we can update the seek position
seekPosition.set(String.valueOf(completedRequestInfo.getTrxLogIndex()));
}
@@ -145,7 +168,8 @@ public void initialize() throws IOException {
public void shutdown() throws IOException {
if (producer != null) {
- producer.close();
+ LOG.info("Closing kafka producer for topic {}", topic);
+ producer.close(Duration.ofSeconds(10));
}
}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerNotificationStrategy.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerNotificationStrategy.java
new file mode 100644
index 00000000000..c9c82bea60d
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerNotificationStrategy.java
@@ -0,0 +1,30 @@
+/*
+ * 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.util.List;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
+
+/**
+ * Interface for strategies which turn completed events into
+ * notifications.
+ */
+public interface OMEventListenerNotificationStrategy {
+
+ List<String> determineEventsForOperation(OmCompletedRequestInfo
completedRequestInfo);
+}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/DateTimeJsonSerializer.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/DateTimeJsonSerializer.java
new file mode 100644
index 00000000000..1e7a1a8bf36
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/DateTimeJsonSerializer.java
@@ -0,0 +1,36 @@
+/*
+ * 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.s3;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import java.io.IOException;
+import java.time.OffsetDateTime;
+import java.time.format.DateTimeFormatter;
+
+/**
+ * A simple replacement for com.amazonaws.internal.DateTimeJsonSerializer.
+ */
+public class DateTimeJsonSerializer extends JsonSerializer<OffsetDateTime> {
+ @Override
+ public void serialize(OffsetDateTime value, JsonGenerator gen,
SerializerProvider provider) throws IOException {
+ // AWS SDK typically uses ISO8601 format for S3 events
+ gen.writeString(value.format(DateTimeFormatter.ISO_OFFSET_DATE_TIME));
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotification.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotification.java
new file mode 100644
index 00000000000..d90b068e8d6
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotification.java
@@ -0,0 +1,560 @@
+/*
+ * 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.s3;
+
+/* copy of
+ * com.amazonaws.services.s3.event.S3EventNotification
+ * class taken from AWS SDK (1.x) with minor changes for build issues
+ * and removed usage of unnecessary AWS specific extension entities:
+ *
+ * - GlacierEventDataEntity
+ * - LifecycleEventDataEntity
+ * - IntelligentTieringEventDataEntity
+ * - ReplicationEventDataEntity
+ *
+ * NOTE: We may not need to fork this class if we can use the SDK one directly
+ * but conversely we may want to make our own customizations.
+ *
+ * Original copyright below:
+ */
+
+/*
+ * 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
+ * limitations under the License.
+ */
+
+//import com.amazonaws.internal.DateTimeJsonSerializer;
+//import com.amazonaws.util.json.Jackson;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+
+/**
+* A helper class that represents a strongly typed S3 EventNotification item
sent
+* to SQS, SNS, or Lambda.
+ *
+ * <p>
+ * <b>Migrating to the AWS SDK for Java v2</b>
+ * <p>
+ * The v2 equivalent of this class is
+ * <a
href="https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/eventnotifications/s3/model/S3EventNotification.html";>S3EventNotification</a>
+ *
+ * <p>
+ * See <a
href="https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration.html";>Migration
Guide</a>
+ * for more information.
+*/
+@SuppressWarnings("checkstyle:all")
+public class S3EventNotification {
+
+ private final List<S3EventNotificationRecord> records;
+
+ /**
+ * Keys for Ozone-specific extensions in the S3 Event schema.
+ */
+ public enum OzoneEventDataKey {
+ IS_DIRECTORY("isDirectory"),
+ IS_RECURSIVE("isRecursive"),
+ IS_OVERWRITE("isOverwrite"),
+ RENAME_FROM_KEY("renameFromKey"),
+ TRX_LOG_INDEX("trxLogIndex"),
+ OP_TYPE("opType");
+
+ private final String jsonKey;
+
+ OzoneEventDataKey(String jsonKey) {
+ this.jsonKey = jsonKey;
+ }
+
+ @Override
+ public String toString() {
+ return jsonKey;
+ }
+ }
+
+ @JsonCreator
+ public S3EventNotification(
+ @JsonProperty(value = "Records") List<S3EventNotificationRecord>
records)
+ {
+ this.records = records;
+ }
+
+ /**
+ * <p>
+ * Parse the JSON string into a S3EventNotification object.
+ * </p>
+ * <p>
+ * The function will try its best to parse input JSON string as best as it
can.
+ * It will not fail even if the JSON string contains unknown properties.
+ * The function will throw SdkClientException if the input JSON string is
+ * not valid JSON.
+ * </p>
+ * @param json
+ * JSON string to parse. Typically this is the body of your SQS
+ * notification message body.
+ *
+ * @return The resulting S3EventNotification object.
+ */
+ //public static S3EventNotification parseJson(String json) {
+ // return Jackson.fromJsonString(json, S3EventNotification.class);
+ //}
+
+ /**
+ * @return the records in this notification
+ */
+ @JsonProperty(value = "Records")
+ public List<S3EventNotificationRecord> getRecords() {
+ return records;
+ }
+
+ //public String toJson() {
+ // return Jackson.toJsonString(this);
+ //}
+
+ public static class UserIdentityEntity {
+
+ private final String principalId;
+
+ @JsonCreator
+ public UserIdentityEntity(
+ @JsonProperty(value = "principalId") String principalId) {
+ this.principalId = principalId;
+ }
+
+ public String getPrincipalId() {
+ return principalId;
+ }
+ }
+
+ public static class S3BucketEntity {
+
+ private final String name;
+ private final UserIdentityEntity ownerIdentity;
+ private final String arn;
+
+ @JsonCreator
+ public S3BucketEntity(
+ @JsonProperty(value = "name") String name,
+ @JsonProperty(value = "ownerIdentity") UserIdentityEntity
ownerIdentity,
+ @JsonProperty(value = "arn") String arn)
+ {
+ this.name = name;
+ this.ownerIdentity = ownerIdentity;
+ this.arn = arn;
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public UserIdentityEntity getOwnerIdentity() {
+ return ownerIdentity;
+ }
+
+ public String getArn() {
+ return arn;
+ }
+ }
+
+ public static class S3ObjectEntity {
+
+ private final String key;
+ private final Long size;
+ private final String eTag;
+ private final String versionId;
+ private final String sequencer;
+
+ @Deprecated
+ public S3ObjectEntity(
+ String key,
+ Integer size,
+ String eTag,
+ String versionId)
+ {
+ this.key = key;
+ this.size = size == null ? null : size.longValue();
+ this.eTag = eTag;
+ this.versionId = versionId;
+ this.sequencer = null;
+ }
+
+ @Deprecated
+ public S3ObjectEntity(
+ String key,
+ Long size,
+ String eTag,
+ String versionId)
+ {
+ this(key, size, eTag, versionId, null);
+ }
+
+ @JsonCreator
+ public S3ObjectEntity(
+ @JsonProperty(value = "key") String key,
+ @JsonProperty(value = "size") Long size,
+ @JsonProperty(value = "eTag") String eTag,
+ @JsonProperty(value = "versionId") String versionId,
+ @JsonProperty(value = "sequencer") String sequencer)
+ {
+ this.key = key;
+ this.size = size;
+ this.eTag = eTag;
+ this.versionId = versionId;
+ this.sequencer = sequencer;
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ /**
+ * S3 URL encodes the key of the object involved in the event. This is
+ * a convenience method to automatically URL decode the key.
+ * @return The URL decoded object key.
+ */
+ //public String getUrlDecodedKey() {
+ // return SdkHttpUtils.urlDecode(getKey());
+ //}
+
+ /**
+ * @deprecated use {@link #getSizeAsLong()} instead.
+ */
+ @Deprecated
+ @JsonIgnore
+ public Integer getSize() {
+ return size == null ? null : size.intValue();
+ }
+
+ @JsonProperty(value = "size")
+ public Long getSizeAsLong() {
+ return size;
+ }
+
+ public String geteTag() {
+ return eTag;
+ }
+
+ public String getVersionId() {
+ return versionId;
+ }
+
+ public String getSequencer() {
+ return sequencer;
+ }
+ }
+
+ public static class S3Entity {
+
+ private final String configurationId;
+ private final S3BucketEntity bucket;
+ private final S3ObjectEntity object;
+ private final String s3SchemaVersion;
+
+ @JsonCreator
+ public S3Entity(
+ @JsonProperty(value = "configurationId") String
configurationId,
+ @JsonProperty(value = "bucket") S3BucketEntity bucket,
+ @JsonProperty(value = "object") S3ObjectEntity object,
+ @JsonProperty(value = "s3SchemaVersion") String
s3SchemaVersion)
+ {
+ this.configurationId = configurationId;
+ this.bucket = bucket;
+ this.object = object;
+ this.s3SchemaVersion = s3SchemaVersion;
+ }
+
+ public String getConfigurationId() {
+ return configurationId;
+ }
+
+ public S3BucketEntity getBucket() {
+ return bucket;
+ }
+
+ public S3ObjectEntity getObject() {
+ return object;
+ }
+
+ public String getS3SchemaVersion() {
+ return s3SchemaVersion;
+ }
+ }
+
+ public static class RequestParametersEntity {
+
+ private final String sourceIPAddress;
+
+ @JsonCreator
+ public RequestParametersEntity(
+ @JsonProperty(value = "sourceIPAddress") String
sourceIPAddress)
+ {
+ this.sourceIPAddress = sourceIPAddress;
+ }
+
+ public String getSourceIPAddress() {
+ return sourceIPAddress;
+ }
+ }
+
+ public static class ResponseElementsEntity {
+
+ private final String xAmzId2;
+ private final String xAmzRequestId;
+
+ @JsonCreator
+ public ResponseElementsEntity(
+ @JsonProperty(value = "x-amz-id-2") String xAmzId2,
+ @JsonProperty(value = "x-amz-request-id") String xAmzRequestId)
+ {
+ this.xAmzId2 = xAmzId2;
+ this.xAmzRequestId = xAmzRequestId;
+ }
+
+ @JsonProperty("x-amz-id-2")
+ public String getxAmzId2() {
+ return xAmzId2;
+ }
+
+ @JsonProperty("x-amz-request-id")
+ public String getxAmzRequestId() {
+ return xAmzRequestId;
+ }
+ }
+
+ public static class LifecycleEventDataEntity {
+
+ private final TransitionEventDataEntity transitionEventData;
+
+ @JsonCreator
+ public LifecycleEventDataEntity(
+ @JsonProperty(value = "transitionEventData")
TransitionEventDataEntity transitionEventData)
+ {
+
+ this.transitionEventData = transitionEventData;
+ }
+
+ public TransitionEventDataEntity getTransitionEventData() {
+ return transitionEventData;
+ }
+ }
+
+ public static class TransitionEventDataEntity {
+ private final String destinationStorageClass;
+
+ @JsonCreator
+ public TransitionEventDataEntity(
+ @JsonProperty("destinationStorageClass") String
destinationStorageClass)
+ {
+ this.destinationStorageClass = destinationStorageClass;
+ }
+
+ public String getDestinationStorageClass() {
+ return destinationStorageClass;
+ }
+ }
+
+ public static class S3EventNotificationRecord {
+
+ private final String awsRegion;
+ private final String eventName;
+ private final String eventSource;
+ private final OffsetDateTime eventTime;
+ private final String eventVersion;
+
+ private final RequestParametersEntity requestParameters;
+ private final ResponseElementsEntity responseElements;
+ private final S3Entity s3;
+ private final UserIdentityEntity userIdentity;
+ private final Map<String, Object> ozoneEventData;
+ //private final GlacierEventDataEntity glacierEventData;
+ //private final LifecycleEventDataEntity lifecycleEventData;
+ //private final IntelligentTieringEventDataEntity
intelligentTieringEventData;
+ //private final ReplicationEventDataEntity replicationEventDataEntity;
+
+ /*
+ @Deprecated
+ public S3EventNotificationRecord(
+ String awsRegion,
+ String eventName,
+ String eventSource,
+ String eventTime,
+ String eventVersion,
+ RequestParametersEntity requestParameters,
+ ResponseElementsEntity responseElements,
+ S3Entity s3,
+ UserIdentityEntity userIdentity)
+ {
+ this(awsRegion,
+ eventName,
+ eventSource,
+ eventTime,
+ eventVersion,
+ requestParameters,
+ responseElements,
+ s3,
+ userIdentity,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ @Deprecated
+ public S3EventNotificationRecord(
+ String awsRegion,
+ String eventName,
+ String eventSource,
+ String eventTime,
+ String eventVersion,
+ RequestParametersEntity requestParameters,
+ ResponseElementsEntity responseElements,
+ S3Entity s3,
+ UserIdentityEntity userIdentity,
+ GlacierEventDataEntity glacierEventData)
+ {
+ this(awsRegion,
+ eventName,
+ eventSource,
+ eventTime,
+ eventVersion,
+ requestParameters,
+ responseElements,
+ s3,
+ userIdentity,
+ glacierEventData,
+ null,
+ null,
+ null);
+ }
+ */
+
+ @JsonCreator
+ public S3EventNotificationRecord(
+ @JsonProperty(value = "awsRegion") String awsRegion,
+ @JsonProperty(value = "eventName") String eventName,
+ @JsonProperty(value = "eventSource") String eventSource,
+ @JsonProperty(value = "eventTime") String eventTime,
+ @JsonProperty(value = "eventVersion") String eventVersion,
+ @JsonProperty(value = "requestParameters")
RequestParametersEntity requestParameters,
+ @JsonProperty(value = "responseElements")
ResponseElementsEntity responseElements,
+ @JsonProperty(value = "s3") S3Entity s3,
+ @JsonProperty(value = "userIdentity") UserIdentityEntity
userIdentity,
+ @JsonProperty(value = "ozoneEventData") Map<String, Object>
ozoneEventData)
+ //@JsonProperty(value = "glacierEventData")
GlacierEventDataEntity glacierEventData,
+ //@JsonProperty(value = "lifecycleEventData")
LifecycleEventDataEntity lifecycleEventData,
+ //@JsonProperty(value = "intelligentTieringEventData")
+ //IntelligentTieringEventDataEntity
intelligentTieringEventData,
+ //@JsonProperty(value = "replicationEventData")
ReplicationEventDataEntity replicationEventData)
+ {
+ this.awsRegion = awsRegion;
+ this.eventName = eventName;
+ this.eventSource = eventSource;
+
+ if (eventTime != null)
+ {
+ this.eventTime = OffsetDateTime.parse(eventTime);
+ } else {
+ this.eventTime = null;
+ }
+
+ this.eventVersion = eventVersion;
+ this.requestParameters = requestParameters;
+ this.responseElements = responseElements;
+ this.s3 = s3;
+ this.userIdentity = userIdentity;
+ this.ozoneEventData = ozoneEventData;
+ //this.glacierEventData = glacierEventData;
+ //this.lifecycleEventData = lifecycleEventData;
+ //this.intelligentTieringEventData = intelligentTieringEventData;
+ //this.replicationEventDataEntity = replicationEventData;
+ }
+
+ public String getAwsRegion() {
+ return awsRegion;
+ }
+
+ public String getEventName() {
+ return eventName;
+ }
+
+ //@JsonIgnore
+ //public S3Event getEventNameAsEnum() {
+ // return S3Event.fromValue(eventName);
+ //}
+
+ public String getEventSource() {
+ return eventSource;
+ }
+
+ @JsonSerialize(using=DateTimeJsonSerializer.class)
+ public OffsetDateTime getEventTime() {
+ return eventTime;
+ }
+
+ public String getEventVersion() {
+ return eventVersion;
+ }
+
+ public RequestParametersEntity getRequestParameters() {
+ return requestParameters;
+ }
+
+ public ResponseElementsEntity getResponseElements() {
+ return responseElements;
+ }
+
+ public S3Entity getS3() {
+ return s3;
+ }
+
+ public UserIdentityEntity getUserIdentity() {
+ return userIdentity;
+ }
+
+ // Ozone extension
+ public Map<String, Object> getOzoneEventData() {
+ return ozoneEventData;
+ }
+
+ //public GlacierEventDataEntity getGlacierEventData() {
+ // return glacierEventData;
+ //}
+
+ //public LifecycleEventDataEntity getLifecycleEventData() { return
lifecycleEventData; }
+
+ //public IntelligentTieringEventDataEntity
getIntelligentTieringEventData() {
+ // return intelligentTieringEventData;
+ //}
+
+ //public ReplicationEventDataEntity getReplicationEventDataEntity() {
return replicationEventDataEntity; }
+
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationBuilder.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationBuilder.java
new file mode 100644
index 00000000000..2f4489fd3f0
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationBuilder.java
@@ -0,0 +1,128 @@
+/*
+ * 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.s3;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.S3BucketEntity;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.S3EventNotificationRecord;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.S3ObjectEntity;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.UserIdentityEntity;
+
+/**
+ * This is a builder for the AWS event notification class
+ * com.amazonaws.services.s3.event.S3EventNotification which is part of
+ * AWS SDK 1.x
+ *
+ * NOTE: the original SDK class is designed primarily for consumer-side
+ * deserialization (parsing incoming JSON). Because it lacks standard public
+ * setters or a fluent API for creation, this builder was added to provide
+ * a clean interface for the serialization path (producing events from Ozone).
+ *
+ * XXX: we may need to fork these classes so that we can customize it to
+ * our needs.
+ */
+public class S3EventNotificationBuilder {
+
+ private static final String REGION = "us-east-1";
+ //private static final String BUCKET_ARN_PREFIX = "arn:aws:s3:" + REGION;
+ private static final String EVENT_SOURCE = "ozone:s3";
+ private static final String EVENT_VERSION = "2.1";
+ private static final String USER_IDENTITY = "some-principalId";
+
+ private static final String SCHEMA_VERSION = "1.0";
+ private static final String CONFIGURATION_ID = "mynotif1";
+
+ private final String objectKey;
+ private final String bucketName;
+ private final String bucketArn;
+ private final String eventName;
+ private final Instant eventTime;
+ private final String etag;
+ private final Map<String, Object> ozoneEventData;
+
+ // mutable fields defaulting to null
+ private Long objectSize;
+ private String objectVersionId;
+ private String objectSequencer;
+
+ public S3EventNotificationBuilder(String objectKey, String bucketName,
String bucketArn, String eventName,
+ Instant eventTime, String etag) {
+ this.objectKey = objectKey;
+ this.bucketName = bucketName;
+ this.bucketArn = bucketArn;
+ this.eventName = eventName;
+ this.eventTime = eventTime;
+ this.etag = etag;
+ this.ozoneEventData = new HashMap<>();
+ }
+
+ public S3EventNotificationBuilder setObjectSize(long objectSize) {
+ this.objectSize = objectSize;
+ return this;
+ }
+
+ public S3EventNotificationBuilder setObjectVersionId(String objectVersionId)
{
+ this.objectVersionId = objectVersionId;
+ return this;
+ }
+
+ public S3EventNotificationBuilder setObjectSequencer(String objectSequencer)
{
+ this.objectSequencer = objectSequencer;
+ return this;
+ }
+
+ public S3EventNotificationBuilder addAllEventData(Map<String, Object>
ozoneEventDataToAdd) {
+ this.ozoneEventData.putAll(ozoneEventDataToAdd);
+ return this;
+ }
+
+ public S3EventNotification build() {
+ UserIdentityEntity userIdentity = new UserIdentityEntity(USER_IDENTITY);
+ S3BucketEntity s3BucketEntity = new S3BucketEntity(bucketName,
userIdentity, bucketArn);
+
+ S3EventNotification.S3ObjectEntity s3ObjectEntity = new S3ObjectEntity(
+ objectKey,
+ objectSize,
+ etag,
+ objectVersionId,
+ objectSequencer);
+
+ S3EventNotification.S3Entity s3Entity = new S3EventNotification.S3Entity(
+ CONFIGURATION_ID,
+ s3BucketEntity,
+ s3ObjectEntity,
+ SCHEMA_VERSION);
+
+ S3EventNotificationRecord eventRecord = new S3EventNotificationRecord(
+ REGION,
+ eventName,
+ EVENT_SOURCE,
+ eventTime.toString(),
+ EVENT_VERSION,
+ new S3EventNotification.RequestParametersEntity(""),
+ new S3EventNotification.ResponseElementsEntity("", ""),
+ s3Entity,
+ new S3EventNotification.UserIdentityEntity(USER_IDENTITY),
+ ozoneEventData);
+
+ return new S3EventNotification(Collections.singletonList(eventRecord));
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationStrategy.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationStrategy.java
new file mode 100644
index 00000000000..9190fbfa3ce
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/S3EventNotificationStrategy.java
@@ -0,0 +1,190 @@
+/*
+ * 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.s3;
+
+import static org.apache.hadoop.ozone.OzoneConsts.OM_KEY_PREFIX;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.commons.lang3.StringUtils;
+import
org.apache.hadoop.ozone.om.eventlistener.OMEventListenerNotificationStrategy;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.OzoneEventDataKey;
+import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * This is a notification strategy to generate events according to S3
+ * notification semantics.
+ */
+public class S3EventNotificationStrategy implements
OMEventListenerNotificationStrategy {
+ public static final Logger LOG =
LoggerFactory.getLogger(S3EventNotificationStrategy.class);
+
+ private static final ObjectMapper MAPPER = new ObjectMapper();
+
+ @Override
+ public List<String> determineEventsForOperation(OmCompletedRequestInfo
requestInfo) {
+
+ switch (requestInfo.getCmdType()) {
+ case CreateVolume:
+ return Collections.singletonList(createS3Event("OzoneVolumeCreated:Put",
+ requestInfo.getVolumeName(), null, null, Collections.emptyMap(),
+ requestInfo.getTrxLogIndex()));
+ case DeleteVolume:
+ return
Collections.singletonList(createS3Event("OzoneVolumeRemoved:Delete",
+ requestInfo.getVolumeName(), null, null, Collections.emptyMap(),
+ requestInfo.getTrxLogIndex()));
+ case CreateBucket:
+ return Collections.singletonList(createS3Event("OzoneBucketCreated:Put",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(), null,
+ Collections.emptyMap(), requestInfo.getTrxLogIndex()));
+ case DeleteBucket:
+ return
Collections.singletonList(createS3Event("OzoneBucketRemoved:Delete",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(), null,
+ Collections.emptyMap(), requestInfo.getTrxLogIndex()));
+ case CreateKey:
+ Map<String, Object> createKeyData = new HashMap<>();
+ createKeyData.put(OzoneEventDataKey.OP_TYPE.toString(), "CreateKey");
+ return Collections.singletonList(createS3Event("ObjectCreated:Put",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
requestInfo.getKeyName(),
+ createKeyData, requestInfo.getTrxLogIndex()));
+ case CommitKey:
+ Map<String, Object> commitKeyData = new HashMap<>();
+ commitKeyData.put(OzoneEventDataKey.OP_TYPE.toString(), "CommitKey");
+ return Collections.singletonList(createS3Event("ObjectCreated:Put",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
requestInfo.getKeyName(),
+ commitKeyData, requestInfo.getTrxLogIndex()));
+ case CreateFile:
+ OmCompletedRequestInfo.OperationArgs.CreateFileArgs createFileArgs
+ = (OmCompletedRequestInfo.OperationArgs.CreateFileArgs)
requestInfo.getOpArgs();
+
+ // XXX: ozoneEventData is an Ozone extension. It is unclear if this
+ // schema makes sense but the general S3 schema is somewhat
+ // freeform. These arguments are more informational than
+ // required so it is unclear as to their necessity.
+ Map<String, Object> createFileEventData = new HashMap<>();
+ createFileEventData.put(OzoneEventDataKey.IS_DIRECTORY.toString(),
false);
+ createFileEventData.put(OzoneEventDataKey.IS_RECURSIVE.toString(),
+ createFileArgs.isRecursive());
+ createFileEventData.put(OzoneEventDataKey.IS_OVERWRITE.toString(),
+ createFileArgs.isOverwrite());
+ createFileEventData.put(OzoneEventDataKey.OP_TYPE.toString(),
"CreateFile");
+
+ return Collections.singletonList(createS3Event("ObjectCreated:Put",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
requestInfo.getKeyName(),
+ createFileEventData, requestInfo.getTrxLogIndex()));
+ case CreateDirectory:
+ // XXX: ozoneEventData is an Ozone extension. It is unclear if this
+ // schema makes sense but the general S3 schema is somewhat
+ // freeform. These arguments are more informational than
+ // required so it is unclear as to their necessity.
+ Map<String, Object> createEventData = new HashMap<>();
+ createEventData.put(OzoneEventDataKey.IS_DIRECTORY.toString(), true);
+ createEventData.put(OzoneEventDataKey.OP_TYPE.toString(),
"CreateDirectory");
+
+ return Collections.singletonList(createS3Event("ObjectCreated:Put",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
requestInfo.getKeyName(),
+ createEventData, requestInfo.getTrxLogIndex()));
+ case DeleteKey:
+ return Collections.singletonList(createS3Event("ObjectRemoved:Delete",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
requestInfo.getKeyName(),
+ Collections.emptyMap(), requestInfo.getTrxLogIndex()));
+ case RenameKey:
+ OmCompletedRequestInfo.OperationArgs.RenameKeyArgs renameKeyArgs
+ = (OmCompletedRequestInfo.OperationArgs.RenameKeyArgs)
requestInfo.getOpArgs();
+
+ String renameFromKey =
S3OzoneEventKeyFormatter.getOzoneKey(requestInfo.getVolumeName(),
+ requestInfo.getBucketName(), requestInfo.getKeyName());
+
+ // XXX: it would be good to be able to convey that this was a
+ // file vs directory rename
+ Map<String, Object> ozoneEventData = new HashMap<>();
+ ozoneEventData.put(OzoneEventDataKey.RENAME_FROM_KEY.toString(),
renameFromKey);
+
+ // NOTE: ObjectRenamed:Rename is an Ozone extension as is the
+ // ozoneEventData map in the S3 event schema.
+ return Collections.singletonList(createS3Event("ObjectRenamed:Rename",
+ requestInfo.getVolumeName(), requestInfo.getBucketName(),
renameKeyArgs.getToKeyName(),
+ ozoneEventData, requestInfo.getTrxLogIndex()));
+ default:
+ LOG.debug("No events for operation {} on {}",
+ requestInfo.getCmdType(),
+ requestInfo.getKeyName());
+ return Collections.emptyList();
+ }
+ }
+
+ static String createS3Event(String eventName, String volumeName, String
bucketName, String keyName,
+ Map<String, Object> ozoneEventData, long trxLogIndex) {
+ try {
+ String objectKey = S3OzoneEventKeyFormatter.getOzoneKey(volumeName,
bucketName, keyName);
+ String bucketArn = (bucketName == null)
+ ? "arn:aws:s3:::" + volumeName
+ : "arn:aws:s3:::" + volumeName + "." + bucketName;
+ Instant eventTime = Instant.now();
+ String etag = UUID.randomUUID().toString();
+
+ Map<String, Object> eventData = new HashMap<>();
+ if (ozoneEventData != null) {
+ eventData.putAll(ozoneEventData);
+ }
+ eventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), trxLogIndex);
+
+ String sequencer = String.format("%016X", trxLogIndex);
+
+ S3EventNotification event =
+ new S3EventNotificationBuilder(objectKey, bucketName, bucketArn,
eventName, eventTime, etag)
+ .setObjectSequencer(sequencer)
+ .addAllEventData(eventData)
+ .build();
+
+ return MAPPER.writer().writeValueAsString(event);
+ } catch (Exception ex) {
+ LOG.error("Failed to create S3 event for {} on {}/{}", eventName,
volumeName, bucketName, ex);
+ return null;
+ }
+ }
+
+ /**
+ * Formats the Ozone key for S3 events.
+ *
+ * NOTE: This differs from OMMetadataManager#getOzoneKey in that it does NOT
+ * include the leading slash, which is standard for S3 notification keys.
+ */
+ private static class S3OzoneEventKeyFormatter {
+ public static String getOzoneKey(String volume, String bucket, String key)
{
+ StringBuilder builder = new StringBuilder();
+ builder.append(volume);
+ if (StringUtils.isNotBlank(bucket)) {
+ builder.append(OM_KEY_PREFIX).append(bucket);
+ if (StringUtils.isNotBlank(key)) {
+ builder.append(OM_KEY_PREFIX);
+ if (!key.equals(OM_KEY_PREFIX)) {
+ builder.append(key);
+ }
+ }
+ }
+ return builder.toString();
+ }
+ }
+}
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/package-info.java
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/package-info.java
new file mode 100644
index 00000000000..78d825a4648
--- /dev/null
+++
b/hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/s3/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * 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 S3 style event generation for the
+ * OM Event Listener.
+ */
+package org.apache.hadoop.ozone.om.eventlistener.s3;
diff --git
a/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerKafkaPublisher.java
b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerKafkaPublisher.java
index 761f8f3c117..76f3de57968 100644
---
a/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerKafkaPublisher.java
+++
b/hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerKafkaPublisher.java
@@ -22,10 +22,18 @@
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import com.fasterxml.jackson.databind.JsonNode;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.hdds.server.JsonUtils;
+import org.apache.hadoop.ozone.OzoneIllegalArgumentException;
+import org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.OzoneEventDataKey;
+import
org.apache.hadoop.ozone.om.eventlistener.s3.S3EventNotification.S3EventNotificationRecord;
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo;
import org.apache.hadoop.ozone.om.helpers.OmCompletedRequestInfo.OperationArgs;
import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.Type;
@@ -49,20 +57,6 @@ public class TestOMEventListenerKafkaPublisher {
@Mock
private OMEventListenerPluginContext pluginContext;
- // helper to create json key/val string for non exhaustive JSON
- // attribute checking
- private static String toJsonKeyVal(String key, String val) {
- return new StringBuilder()
- .append('\"')
- .append(key)
- .append('\"')
- .append(':')
- .append('\"')
- .append(val)
- .append('\"')
- .toString();
- }
-
private static OmCompletedRequestInfo buildCompletedRequestInfo(
long trxLogIndex, Type cmdType, String keyName, OperationArgs
opArgs) {
@@ -101,17 +95,50 @@ private List<String>
captureEventsProducedByOperation(OmCompletedRequestInfo op,
return events;
}
+ private S3EventNotificationRecord getFirstRecord(List<String> events) throws
IOException {
+ assertThat(events).hasSize(1);
+ S3EventNotification notification = JsonUtils.getDefaultMapper()
+ .readValue(events.get(0), S3EventNotification.class);
+ assertThat(notification.getRecords()).hasSize(1);
+ return notification.getRecords().get(0);
+ }
+
@Test
public void testCreateKeyRequestProducesS3CreatedEvent() throws
InterruptedException, IOException {
OmCompletedRequestInfo createRequest = buildCompletedRequestInfo(1L,
Type.CreateKey, "some/key1",
new OperationArgs.NoArgs());
List<String> events = captureEventsProducedByOperation(createRequest, 1);
- assertThat(events).hasSize(1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("ObjectCreated:Put");
+
assertThat(record.getS3().getObject().getKey()).isEqualTo("vol1/bucket1/some/key1");
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
1L));
+
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.OP_TYPE.toString(), "CreateKey");
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 1);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
+ }
+
+ @Test
+ public void testCommitKeyRequestProducesS3CreatedEvent() throws
InterruptedException, IOException {
+ OmCompletedRequestInfo commitRequest = buildCompletedRequestInfo(8L,
Type.CommitKey, "some/key1_commit",
+ new OperationArgs.NoArgs());
+
+ List<String> events = captureEventsProducedByOperation(commitRequest, 1);
+ S3EventNotificationRecord record = getFirstRecord(events);
- assertThat(events.get(0))
- .contains(toJsonKeyVal("key", "vol1/bucket1/some/key1"))
- .contains(toJsonKeyVal("type", "CreateKey"));
+ assertThat(record.getEventName()).isEqualTo("ObjectCreated:Put");
+
assertThat(record.getS3().getObject().getKey()).isEqualTo("vol1/bucket1/some/key1_commit");
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
8L));
+
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.OP_TYPE.toString(), "CommitKey");
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 8);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
}
@Test
@@ -123,11 +150,20 @@ public void testCreateFileRequestProducesS3CreatedEvent()
throws InterruptedExce
new OperationArgs.CreateFileArgs(recursive, overwrite));
List<String> events = captureEventsProducedByOperation(createRequest, 1);
- assertThat(events).hasSize(1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("ObjectCreated:Put");
+
assertThat(record.getS3().getObject().getKey()).isEqualTo("vol1/bucket1/some/key2");
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
2L));
- assertThat(events.get(0))
- .contains(toJsonKeyVal("key", "vol1/bucket1/some/key2"))
- .contains(toJsonKeyVal("type", "CreateFile"));
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.IS_DIRECTORY.toString(), false);
+ expectedEventData.put(OzoneEventDataKey.IS_RECURSIVE.toString(), false);
+ expectedEventData.put(OzoneEventDataKey.IS_OVERWRITE.toString(), true);
+ expectedEventData.put(OzoneEventDataKey.OP_TYPE.toString(), "CreateFile");
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 2);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
}
@Test
@@ -136,23 +172,138 @@ public void
testCreateDirectoryRequestProducesS3CreatedEvent() throws Interrupte
new OperationArgs.NoArgs());
List<String> events = captureEventsProducedByOperation(createRequest, 1);
- assertThat(events).hasSize(1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("ObjectCreated:Put");
+
assertThat(record.getS3().getObject().getKey()).isEqualTo("vol1/bucket1/some/key3");
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
3L));
- assertThat(events.get(0))
- .contains(toJsonKeyVal("key", "vol1/bucket1/some/key3"))
- .contains(toJsonKeyVal("type", "CreateDirectory"));
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.IS_DIRECTORY.toString(), true);
+ expectedEventData.put(OzoneEventDataKey.OP_TYPE.toString(),
"CreateDirectory");
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 3);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
}
@Test
- public void testRenameRequestProducesRenameKeyEvent() throws
InterruptedException, IOException {
+ public void testRenameRequestProducesS3RenamedEvent() throws
InterruptedException, IOException {
OmCompletedRequestInfo renameRequest = buildCompletedRequestInfo(4L,
Type.RenameKey, "some/key4",
new OperationArgs.RenameKeyArgs("some/key_RENAMED"));
List<String> events = captureEventsProducedByOperation(renameRequest, 1);
- assertThat(events).hasSize(1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("ObjectRenamed:Rename");
+
assertThat(record.getS3().getObject().getKey()).isEqualTo("vol1/bucket1/some/key_RENAMED");
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
4L));
+
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.RENAME_FROM_KEY.toString(),
"vol1/bucket1/some/key4");
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 4);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
+ }
+
+ @Test
+ public void testCreateVolumeRequestProducesOzoneVolumeCreatedEvent() throws
IOException {
+ OmCompletedRequestInfo createRequest = new OmCompletedRequestInfo.Builder()
+ .setTrxLogIndex(6L)
+ .setCmdType(Type.CreateVolume)
+ .setVolumeName(VOLUME_NAME)
+ .setBucketName(null)
+ .setKeyName(null)
+ .setCreationTime(Time.now())
+ .setOpArgs(new OperationArgs.NoArgs())
+ .build();
+
+ List<String> events = captureEventsProducedByOperation(createRequest, 1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("OzoneVolumeCreated:Put");
+ assertThat(record.getS3().getBucket().getName()).isNull();
+ assertThat(record.getS3().getObject().getKey()).isEqualTo(VOLUME_NAME);
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
6L));
+
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 6);
- assertThat(events.get(0))
- .contains(toJsonKeyVal("key", "vol1/bucket1/some/key4"))
- .contains(toJsonKeyVal("type", "RenameKey"));
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
+ }
+
+ @Test
+ public void testCreateBucketRequestProducesOzoneBucketCreatedEvent() throws
IOException {
+ OmCompletedRequestInfo createRequest = new OmCompletedRequestInfo.Builder()
+ .setTrxLogIndex(7L)
+ .setCmdType(Type.CreateBucket)
+ .setVolumeName(VOLUME_NAME)
+ .setBucketName(BUCKET_NAME)
+ .setKeyName(null)
+ .setCreationTime(Time.now())
+ .setOpArgs(new OperationArgs.NoArgs())
+ .build();
+
+ List<String> events = captureEventsProducedByOperation(createRequest, 1);
+ S3EventNotificationRecord record = getFirstRecord(events);
+
+ assertThat(record.getEventName()).isEqualTo("OzoneBucketCreated:Put");
+ assertThat(record.getS3().getBucket().getName()).isEqualTo(BUCKET_NAME);
+ assertThat(record.getS3().getObject().getKey()).isEqualTo(VOLUME_NAME +
"/" + BUCKET_NAME);
+
assertThat(record.getS3().getObject().getSequencer()).isEqualTo(String.format("%016X",
7L));
+
+ Map<String, Object> expectedEventData = new HashMap<>();
+ expectedEventData.put(OzoneEventDataKey.TRX_LOG_INDEX.toString(), 7);
+
+ assertThat(record.getOzoneEventData()).isEqualTo(expectedEventData);
+ }
+
+ @Test
+ public void testEventDateFormatIsIso8601() throws IOException {
+ OmCompletedRequestInfo createRequest = buildCompletedRequestInfo(5L,
Type.CreateKey, "date/test",
+ new OperationArgs.NoArgs());
+
+ List<String> events = captureEventsProducedByOperation(createRequest, 1);
+ String rawJson = events.get(0);
+
+ // Parse raw JSON to extract the eventTime string exactly as it appears in
the message
+ JsonNode root = JsonUtils.readTree(rawJson);
+ String eventTimeStr =
root.path("Records").get(0).path("eventTime").asText();
+
+ // Validate that it matches ISO-8601 offset format (e.g.,
2026-05-20T13:45:00Z or +01:00)
+ // This ensures our DateTimeJsonSerializer is working as expected.
+
assertThat(eventTimeStr).matches("^\\d{4}-\\d{2}-\\d{2}T\\d{2}:\\d{2}:\\d{2}.*");
+
+ // Also ensure it is actually parsable by the standard Java 8 API
+ java.time.OffsetDateTime.parse(eventTimeStr);
+ }
+
+ @Test
+ public void testInitializeFailsWhenStrategyInstantiationThrows() {
+ OzoneConfiguration conf = new OzoneConfiguration();
+ conf.setClass("ozone.om.plugin.kafka.notification.strategy",
+ ThrowingStrategy.class, OMEventListenerNotificationStrategy.class);
+
+ OMEventListenerKafkaPublisher plugin = new OMEventListenerKafkaPublisher();
+ try (MockedConstruction<OMEventListenerKafkaPublisher.KafkaClientWrapper>
mockedKafkaClientWrapper =
+
mockConstruction(OMEventListenerKafkaPublisher.KafkaClientWrapper.class)) {
+
+
org.junit.jupiter.api.Assertions.assertThrows(OzoneIllegalArgumentException.class,
() -> {
+ plugin.initialize(conf, pluginContext);
+ });
+ }
+ }
+
+ /**
+ * A mock strategy that throws an exception during instantiation for testing.
+ */
+ public static class ThrowingStrategy implements
OMEventListenerNotificationStrategy {
+ public ThrowingStrategy() {
+ throw new RuntimeException("Simulated instantiation failure");
+ }
+
+ @Override
+ public List<String> determineEventsForOperation(OmCompletedRequestInfo
completedRequestInfo) {
+ return null;
+ }
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
