HDDS-16014. [Ozone Versioning] Enabled Write Path

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConsts.java

@@ -299,6 +299,7 @@ public final class OzoneConsts {
   public static final String STORAGE_TYPE = "storageType";
   public static final String RESOURCE_TYPE = "resourceType";
   public static final String IS_VERSION_ENABLED = "isVersionEnabled";
+  public static final String VERSIONING_STATUS = "versioningStatus";
   public static final String CREATION_TIME = "creationTime";
   public static final String MODIFICATION_TIME = "modificationTime";
   public static final String DATA_SIZE = "dataSize";

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -5163,4 +5163,18 @@
     <tag>OZONE, RATIS, OM</tag>
     <description>The maximum number of events that can be pending in OM Ratis.</description>
   </property>
+  <property>
+    <name>ozone.om.versioning.version-id-generator</name>
+    <value>org.apache.hadoop.ozone.om.helpers.TransactionIndexVersionIdGenerator</value>
+    <tag>OZONE, OM</tag>
+    <description>
+      Implementation of org.apache.hadoop.ozone.om.helpers.VersionIdGenerator used to assign the
+      versionId of an object version. The default uses the index of the committing transaction;
+      PinnedFirstVersionIdGenerator additionally pins the first version of every key to a reserved
+      sentinel, so that it can be referenced without listing the key's versions first.
+      The setting is cluster-wide and may be changed on a running cluster; a write whose generated
+      versionId already exists on the key is rejected, and the existing version has to be deleted
+      before that id can be written again.
+    </description>
+  </property>
 </configuration>

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

@@ -21,6 +21,8 @@
 import org.apache.hadoop.hdds.client.ReplicationFactor;
 import org.apache.hadoop.hdds.client.ReplicationType;
 import org.apache.hadoop.ozone.om.helpers.BucketLayout;
+import org.apache.hadoop.ozone.om.helpers.TransactionIndexVersionIdGenerator;
+import org.apache.hadoop.ozone.om.helpers.VersionIdGenerator;
 import org.apache.ratis.util.TimeDuration;
 
 /**
@@ -715,6 +717,11 @@ public final class OMConfigKeys {
       "ozone.om.ratis.events.max.limit";
   public static final int OZONE_OM_RATIS_EVENTS_MAX_LIMIT_DEFAULT = 100;
 
+  public static final String OZONE_OM_VERSIONING_VERSION_ID_GENERATOR =
+      "ozone.om.versioning.version-id-generator";
+  public static final Class<? extends VersionIdGenerator>
+      OZONE_OM_VERSIONING_VERSION_ID_GENERATOR_DEFAULT = TransactionIndexVersionIdGenerator.class;
+
   /**
    * Never constructed.
    */

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

@@ -0,0 +1,73 @@
+/*
+ * 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.helpers;
+
+import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.BucketVersioningStatusProto;
+
+/**
+ * S3-compatible bucket versioning state machine:
+ * UNVERSIONED -> ENABLED <-> SUSPENDED.
+ * Once versioning has been enabled, a bucket can never return to UNVERSIONED.
+ */
+public enum BucketVersioningStatus {
+  UNVERSIONED,
+  ENABLED,
+  SUSPENDED;
+
+  public static BucketVersioningStatus fromProto(BucketVersioningStatusProto proto) {
+    switch (proto) {
+    case VERSIONING_ENABLED:
+      return ENABLED;
+    case VERSIONING_SUSPENDED:
+      return SUSPENDED;
+    case UNVERSIONED:
+    default:
+      return UNVERSIONED;
+    }
+  }
+
+  public BucketVersioningStatusProto toProto() {
+    switch (this) {
+    case ENABLED:
+      return BucketVersioningStatusProto.VERSIONING_ENABLED;
+    case SUSPENDED:
+      return BucketVersioningStatusProto.VERSIONING_SUSPENDED;
+    default:
+      return BucketVersioningStatusProto.UNVERSIONED;
+    }
+  }
+
+  /** Maps the legacy isVersionEnabled flag of buckets without an explicit status. */
+  public static BucketVersioningStatus fromVersionEnabledFlag(boolean isVersionEnabled) {
+    return isVersionEnabled ? ENABLED : UNVERSIONED;
+  }
+
+  /** The legacy isVersionEnabled flag value kept in sync with this status. */
+  public boolean toVersionEnabledFlag() {
+    return this == ENABLED;
+  }
+
+  /**
+   * Whether a bucket may transition from this status to {@code target}.
+   * The only forbidden transition is back to UNVERSIONED after versioning
+   * has been enabled or suspended (matches the S3 state machine).
+   */
+  public boolean canTransitionTo(BucketVersioningStatus target) {
+    return this == UNVERSIONED || target != UNVERSIONED;
+  }
+}

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

@@ -44,6 +44,11 @@ public final class OmBucketArgs extends WithMetadata implements Auditable {
    * Bucket Version flag.
    */
   private final Boolean isVersionEnabled;
+  /**
+   * S3-compatible versioning status; null when not being changed.
+   * Takes precedence over isVersionEnabled when both are set.
+   */
+  private final BucketVersioningStatus versioningStatus;
   /**
    * Type of storage to be used for this bucket.
    * [RAM_DISK, SSD, DISK, ARCHIVE]
@@ -73,6 +78,7 @@ private OmBucketArgs(Builder b) {
     this.volumeName = b.volumeName;
     this.bucketName = b.bucketName;
     this.isVersionEnabled = b.isVersionEnabled;
+    this.versioningStatus = b.versioningStatus;
     this.storageType = b.storageType;
     this.ownerName = b.ownerName;
     this.defaultReplicationConfig = b.defaultReplicationConfig;
@@ -108,6 +114,14 @@ public Boolean getIsVersionEnabled() {
     return isVersionEnabled;
   }
 
+  /**
+   * Returns the requested versioning status, or null if not being changed.
+   * @return BucketVersioningStatus
+   */
+  public BucketVersioningStatus getVersioningStatus() {
+    return versioningStatus;
+  }
+
   /**
    * Returns the type of storage to be used.
    * @return StorageType
@@ -190,6 +204,9 @@ public Map<String, String> toAuditMap() {
         getMetadata().get(OzoneConsts.GDPR_FLAG));
     auditMap.put(OzoneConsts.IS_VERSION_ENABLED,
                 String.valueOf(this.isVersionEnabled));
+    if (this.versioningStatus != null) {
+      auditMap.put(OzoneConsts.VERSIONING_STATUS, this.versioningStatus.name());
+    }
     if (this.storageType != null) {
       auditMap.put(OzoneConsts.STORAGE_TYPE, this.storageType.name());
     }
@@ -227,6 +244,7 @@ public static class Builder extends WithMetadata.Builder {
     private String volumeName;
     private String bucketName;
     private Boolean isVersionEnabled;
+    private BucketVersioningStatus versioningStatus;
     private StorageType storageType;
     private boolean quotaInBytesSet = false;
     private long quotaInBytes;
@@ -261,6 +279,11 @@ public Builder setIsVersionEnabled(Boolean versionFlag) {
       return this;
     }
 
+    public Builder setVersioningStatus(BucketVersioningStatus status) {
+      this.versioningStatus = status;
+      return this;
+    }
+
     @Deprecated
     public Builder setBucketEncryptionKey(BucketEncryptionKeyInfo info) {
       if (info == null || info.getKeyName() != null) {
@@ -339,6 +362,9 @@ public BucketArgs getProtobuf() {
     if (isVersionEnabled != null) {
       builder.setIsVersionEnabled(isVersionEnabled);
     }
+    if (versioningStatus != null) {
+      builder.setVersioningStatus(versioningStatus.toProto());
+    }
     if (storageType != null) {
       builder.setStorageType(storageType.toProto());
     }
@@ -381,6 +407,10 @@ public static Builder builderFromProtobuf(BucketArgs bucketArgs) {
     if (bucketArgs.hasIsVersionEnabled()) {
       builder.setIsVersionEnabled(bucketArgs.getIsVersionEnabled());
     }
+    if (bucketArgs.hasVersioningStatus()) {
+      builder.setVersioningStatus(
+          BucketVersioningStatus.fromProto(bucketArgs.getVersioningStatus()));
+    }
     if (bucketArgs.hasStorageType()) {
       builder.setStorageType(StorageType.valueOf(bucketArgs.getStorageType()));
     }

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

@@ -59,9 +59,13 @@ public final class OmBucketInfo extends WithObjectID implements Auditable, CopyO
    */
   private final ImmutableList<OzoneAcl> acls;
   /**
-   * Bucket Version flag.
+   * Bucket Version flag, kept in sync with versioningStatus (ENABLED -> true).
    */
   private final boolean isVersionEnabled;
+  /**
+   * S3-compatible versioning status; authoritative over isVersionEnabled.
+   */
+  private final BucketVersioningStatus versioningStatus;
   /**
    * Type of storage to be used for this bucket.
    * [RAM_DISK, SSD, DISK, ARCHIVE]
@@ -118,7 +122,9 @@ private OmBucketInfo(Builder b) {
     this.volumeName = b.volumeName;
     this.bucketName = b.bucketName;
     this.acls = b.acls.build();
-    this.isVersionEnabled = b.isVersionEnabled;
+    this.versioningStatus = b.versioningStatus != null ? b.versioningStatus
+        : BucketVersioningStatus.fromVersionEnabledFlag(b.isVersionEnabled);
+    this.isVersionEnabled = this.versioningStatus.toVersionEnabledFlag();
     this.storageType = b.storageType;
     this.creationTime = b.creationTime;
     this.modificationTime = b.modificationTime;
@@ -173,6 +179,27 @@ public boolean getIsVersionEnabled() {
     return isVersionEnabled;
   }
 
+  /**
+   * Returns the S3-compatible versioning status; never null.
+   * @return BucketVersioningStatus
+   */
+  public BucketVersioningStatus getVersioningStatus() {
+    return versioningStatus;
+  }
+
+  /**
+   * Whether S3-compatible versioning is in effect, i.e. whether a write keeps
+   * the previous current version as a separate record in the versionedKeyTable.
+   * True for status ENABLED on an OBJECT_STORE bucket; buckets of other layouts
+   * carrying the legacy isVersionEnabled flag keep the legacy in-record block
+   * version behaviour.
+   * @return whether writes create versionedKeyTable records
+   */
+  public boolean isS3VersioningEnabled() {
+    return versioningStatus == BucketVersioningStatus.ENABLED
+        && bucketLayout == BucketLayout.OBJECT_STORE;
+  }
+
   /**
    * Returns the type of storage to be used.
    * @return StorageType
@@ -338,6 +365,7 @@ public Map<String, String> toAuditMap() {
         (this.acls != null) ? this.acls.toString() : null);
     auditMap.put(OzoneConsts.IS_VERSION_ENABLED,
         String.valueOf(this.isVersionEnabled));
+    auditMap.put(OzoneConsts.VERSIONING_STATUS, this.versioningStatus.name());
     auditMap.put(OzoneConsts.STORAGE_TYPE,
         (this.storageType != null) ? this.storageType.name() : null);
     auditMap.put(OzoneConsts.CREATION_TIME, String.valueOf(this.creationTime));
@@ -379,6 +407,7 @@ public Builder toBuilder() {
         .setBucketName(bucketName)
         .setStorageType(storageType)
         .setIsVersionEnabled(isVersionEnabled)
+        .setVersioningStatus(versioningStatus)
         .setCreationTime(creationTime)
         .setModificationTime(modificationTime)
         .setBucketEncryptionKey(bekInfo)
@@ -404,6 +433,7 @@ public static class Builder extends WithObjectID.Builder<OmBucketInfo> {
     private String bucketName;
     private final AclListBuilder acls;
     private boolean isVersionEnabled;
+    private BucketVersioningStatus versioningStatus;
     private StorageType storageType = StorageType.DISK;
     private long creationTime;
     private long modificationTime;
@@ -462,6 +492,22 @@ public Builder addAcl(OzoneAcl ozoneAcl) {
 
     public Builder setIsVersionEnabled(boolean versionFlag) {
       this.isVersionEnabled = versionFlag;
+      // Keep versioningStatus in sync for callers that only know the legacy
+      // flag; an explicitly SUSPENDED status is preserved on disable.
+      if (versionFlag) {
+        this.versioningStatus = BucketVersioningStatus.ENABLED;
+      } else if (versioningStatus != BucketVersioningStatus.SUSPENDED) {
+        this.versioningStatus = BucketVersioningStatus.UNVERSIONED;
+      }
+      return this;
+    }
+
+    /** No-op when status is null (e.g. records without the new field). */
+    public Builder setVersioningStatus(BucketVersioningStatus status) {
+      if (status != null) {
+        this.versioningStatus = status;
+        this.isVersionEnabled = status.toVersionEnabledFlag();
+      }
       return this;
     }
 
@@ -600,6 +646,7 @@ public BucketInfo getProtobuf() {
         .setBucketName(bucketName)
         .addAllAcls(OzoneAclUtil.toProtobuf(acls))
         .setIsVersionEnabled(isVersionEnabled)
+        .setVersioningStatus(versioningStatus.toProto())
         .setStorageType(storageType.toProto())
         .setCreationTime(creationTime)
         .setModificationTime(modificationTime)
@@ -657,6 +704,9 @@ public static Builder builderFromProtobuf(BucketInfo bucketInfo,
         .setAcls(bucketInfo.getAclsList().stream().map(
             OzoneAcl::fromProtobuf).collect(Collectors.toList()))
         .setIsVersionEnabled(bucketInfo.getIsVersionEnabled())
+        .setVersioningStatus(bucketInfo.hasVersioningStatus()
+            ? BucketVersioningStatus.fromProto(bucketInfo.getVersioningStatus())
+            : null)
         .setStorageType(StorageType.valueOf(bucketInfo.getStorageType()))
         .setCreationTime(bucketInfo.getCreationTime())
         .setUsedBytes(bucketInfo.getUsedBytes())
@@ -763,6 +813,7 @@ public boolean equals(Object o) {
         bucketName.equals(that.bucketName) &&
         Objects.equals(acls, that.acls) &&
         Objects.equals(isVersionEnabled, that.isVersionEnabled) &&
+        versioningStatus == that.versioningStatus &&
         storageType == that.storageType &&
         getObjectID() == that.getObjectID() &&
         getUpdateID() == that.getUpdateID() &&
@@ -791,6 +842,7 @@ public String toString() {
         ", bucketName='" + bucketName + "'" +
         ", acls=" + acls +
         ", isVersionEnabled=" + isVersionEnabled +
+        ", versioningStatus=" + versioningStatus +
         ", storageType=" + storageType +
         ", creationTime=" + creationTime +
         ", bekInfo=" + bekInfo +