HDDS-15645. ListObjectsV2 returns Owner when FetchOwner is not requested

hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java

@@ -1667,6 +1667,30 @@ public void testListObjectsV2BucketNotExist() {
     assertEquals("NoSuchBucket", ase.getErrorCode());
   }
 
+  @Test
+  public void testListObjectsV2FetchOwner() {
+    final String bucketName = getBucketName("fetch-owner");
+    final String keyName = getKeyName("obj");
+    s3Client.createBucket(bucketName);
+    s3Client.putObject(bucketName, keyName, RandomStringUtils.secure().nextAlphanumeric(5));
+
+    ListObjectsV2Result defaultResponse = s3Client.listObjectsV2(
+        new ListObjectsV2Request().withBucketName(bucketName));
+    assertThat(defaultResponse.getObjectSummaries()).isNotEmpty();
+    assertNull(defaultResponse.getObjectSummaries().get(0).getOwner());
+
+    ListObjectsV2Result falseResponse = s3Client.listObjectsV2(
+        new ListObjectsV2Request().withBucketName(bucketName).withFetchOwner(false));
+    assertNull(falseResponse.getObjectSummaries().get(0).getOwner());
+
+    ListObjectsV2Result trueResponse = s3Client.listObjectsV2(
+        new ListObjectsV2Request().withBucketName(bucketName).withFetchOwner(true));
+    Owner owner = trueResponse.getObjectSummaries().get(0).getOwner();
+    assertNotNull(owner);
+    assertNotNull(owner.getDisplayName());
+    assertEquals(S3Owner.DEFAULT_S3OWNER_ID, owner.getId());
+  }
+
   @Test
   public void testHighLevelMultipartUpload(@TempDir Path tempDir) throws Exception {
     TransferManager tm = TransferManagerBuilder.standard()

hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v2/AbstractS3SDKV2Tests.java

@@ -1109,6 +1109,30 @@ public void testListObjectsSpecialKeyNamesV2() throws Exception {
     assertEquals(S3SDKTestUtils.S3_SPECIAL_KEY_NAMES, listedKeys);
   }
 
+  @Test
+  public void testListObjectsV2FetchOwner() {
+    final String bucketName = getBucketName("fetch-owner");
+    final String keyName = getKeyName("obj");
+    s3Client.createBucket(b -> b.bucket(bucketName));
+    s3Client.putObject(b -> b.bucket(bucketName).key(keyName),
+        RequestBody.fromString("x"));
+
+    ListObjectsV2Response defaultResponse = s3Client.listObjectsV2(
+        ListObjectsV2Request.builder().bucket(bucketName).build());
+    assertThat(defaultResponse.contents()).isNotEmpty();
+    assertNull(defaultResponse.contents().get(0).owner());
+
+    ListObjectsV2Response falseResponse = s3Client.listObjectsV2(
+        ListObjectsV2Request.builder().bucket(bucketName).fetchOwner(false).build());
+    assertNull(falseResponse.contents().get(0).owner());
+
+    ListObjectsV2Response trueResponse = s3Client.listObjectsV2(
+        ListObjectsV2Request.builder().bucket(bucketName).fetchOwner(true).build());
+    assertNotNull(trueResponse.contents().get(0).owner());
+    assertNotNull(trueResponse.contents().get(0).owner().displayName());
+    assertEquals(S3Owner.DEFAULT_S3OWNER_ID, trueResponse.contents().get(0).owner().id());
+  }
+
   private void testListObjectsMany(boolean isListV2) throws Exception {
     final String bucketName = getBucketName();
     s3Client.createBucket(b -> b.bucket(bucketName));

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/commontypes/RequestParameters.java

@@ -65,6 +65,20 @@ default void setInt(String key, int value) {
     }
   }
 
+  default boolean getBoolean(String key, boolean defaultValue) {
+    final String value = get(key);
+    if (value == null) {
+      return defaultValue;
+    }
+    if ("true".equalsIgnoreCase(value)) {
+      return true;
+    }
+    if ("false".equalsIgnoreCase(value)) {
+      return false;
+    }
+    throw S3ErrorTable.newError(S3ErrorTable.INVALID_ARGUMENT, key);
+  }
+
   /** Mutable implementation based on {@link MultivaluedMap}. */
   final class MultivaluedMapImpl implements Mutable {
     private final MultivaluedMap<String, String> params;

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/endpoint/BucketEndpoint.java

@@ -113,7 +113,7 @@ Response handleGetRequest(S3RequestContext context, String bucketName) throws IO
     int maxKeys = queryParams().getInt(QueryParams.MAX_KEYS, 1000);
     String prefix = queryParams().get(QueryParams.PREFIX, "");
     String startAfter = queryParams().get(QueryParams.START_AFTER);
-
+    boolean includeOwner = shouldIncludeOwnerInListResponse();
     Iterator<? extends OzoneKey> ozoneKeyIterator = null;
     // AWS S3 treats an empty continuation-token as no token: list from the
     // start and echo the empty token back (see setContinueToken below).
@@ -217,11 +217,11 @@ Response handleGetRequest(S3RequestContext context, String bucketName) throws IO
           } else {
             // means our key is matched with prefix if prefix is given and it
             // does not have any common prefix.
-            addKey(response, next);
+            addKey(response, next, includeOwner);
             count++;
           }
         } else {
-          addKey(response, next);
+          addKey(response, next, includeOwner);
           count++;
         }
 
@@ -402,7 +402,7 @@ public MultiDeleteResponse multiDelete(
     return result;
   }
 
-  private void addKey(ListObjectResponse response, OzoneKey next) {
+  private void addKey(ListObjectResponse response, OzoneKey next, boolean includeOwner) {
     KeyMetadata keyMetadata = new KeyMetadata();
     keyMetadata.setKey(EncodingTypeObject.createNullable(next.getName(),
         response.getEncodingType()));
@@ -414,8 +414,9 @@ private void addKey(ListObjectResponse response, OzoneKey next) {
     keyMetadata.setStorageClass(S3StorageType.fromReplicationConfig(
         next.getReplicationConfig()).toString());
     keyMetadata.setLastModified(next.getModificationTime());
-    String displayName = next.getOwner();
-    keyMetadata.setOwner(S3Owner.of(displayName));
+    if (includeOwner) {
+      keyMetadata.setOwner(S3Owner.of(next.getOwner()));
+    }
     response.addKey(keyMetadata);
   }
 
@@ -439,4 +440,10 @@ protected void init() {
         .build();
     handler = new AuditingBucketOperationHandler(chain);
   }
+
+  private boolean shouldIncludeOwnerInListResponse() {
+    int listType = queryParams().getInt(QueryParams.LIST_TYPE, 1);
+    boolean fetchOwner = queryParams().getBoolean(QueryParams.FETCH_OWNER, false);
+    return listType != 2 || fetchOwner;
+  }
 }

hadoop-ozone/s3gateway/src/main/java/org/apache/hadoop/ozone/s3/util/S3Consts.java

@@ -144,6 +144,8 @@ public static final class QueryParams {
     public static final String DELIMITER = "delimiter";
     public static final String ENCODING_TYPE = "encoding-type";
     public static final String KEY_MARKER = "key-marker";
+    public static final String FETCH_OWNER = "fetch-owner";
+    public static final String LIST_TYPE = "list-type";
     // GetBucketLocation is not implemented
     public static final String LOCATION = "location";
     public static final String MARKER = "marker";

hadoop-ozone/s3gateway/src/test/java/org/apache/hadoop/ozone/s3/endpoint/TestBucketList.java

@@ -700,6 +700,43 @@ public void continuationTokenXmlElementName() throws Exception {
         "response must not use the non-AWS <continueToken> element");
   }
 
+  @Test
+  public void listObjectOwnerOmittedForListV2ByDefault() throws OS3Exception, IOException {
+    OzoneClient client = createClientWithKeys("key1", "key2");
+    BucketEndpoint endpoint = newBucketEndpointBuilder().setClient(client).build();
+
+    endpoint.queryParamsForTest().setInt(QueryParams.LIST_TYPE, 2);
+    ListObjectResponse response = (ListObjectResponse) endpoint.get("b1").getEntity();
+
+    assertEquals(2, response.getContents().size());
+    assertNull(response.getContents().get(0).getOwner());
+    assertNull(response.getContents().get(1).getOwner());
+  }
+
+  @Test
+  public void listObjectOwnerOmittedForListV2WhenFetchOwnerFalse() throws OS3Exception, IOException {
+    OzoneClient client = createClientWithKeys("key1");
+    BucketEndpoint endpoint = newBucketEndpointBuilder().setClient(client).build();
+
+    endpoint.queryParamsForTest().setInt(QueryParams.LIST_TYPE, 2);
+    endpoint.queryParamsForTest().set(QueryParams.FETCH_OWNER, "false");
+    ListObjectResponse response = (ListObjectResponse) endpoint.get("b1").getEntity();
+
+    assertNull(response.getContents().get(0).getOwner());
+  }
+
+  @Test
+  public void listObjectOwnerIncludedForListV2WhenFetchOwnerTrue() throws OS3Exception, IOException {
+    OzoneClient client = createClientWithKeys("key1");
+    BucketEndpoint endpoint = newBucketEndpointBuilder().setClient(client).build();
+
+    endpoint.queryParamsForTest().setInt(QueryParams.LIST_TYPE, 2);
+    endpoint.queryParamsForTest().set(QueryParams.FETCH_OWNER, "true");
+    ListObjectResponse response = (ListObjectResponse) endpoint.get("b1").getEntity();
+
+    assertNotNull(response.getContents().get(0).getOwner());
+  }
+
   private OzoneClient createClientWithKeys(String... keys) throws IOException {
     OzoneClient client = new OzoneClientStub();