environment) {
- Credentials credentials = AbstractS3DelegationTokenReceiver.getCredentials();
+ S3SessionCredentials credentials = AbstractS3DelegationTokenReceiver.getCredentials();
if (credentials != null) {
maybeSetEnvironmentVariable(
environment, "AWS_ACCESS_KEY_ID", credentials.getAccessKeyId());
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProvider.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProvider.java
index 82e99dcde89568..fd1f02d29e2964 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProvider.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProvider.java
@@ -25,18 +25,19 @@
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.StringUtils;
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
-import com.amazonaws.auth.BasicAWSCredentials;
-import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
-import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceClientBuilder;
-import com.amazonaws.services.securitytoken.model.Credentials;
-import com.amazonaws.services.securitytoken.model.GetSessionTokenResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Optional;
-/** Delegation token provider for S3 filesystems. */
+/**
+ * Delegation token provider for S3 filesystems.
+ *
+ * This class deliberately references no AWS SDK types (see {@link S3SessionCredentials}): the
+ * STS call to obtain session credentials is left to {@link #getSessionCredentials(String, String,
+ * String)}, implemented with AWS SDK v1 in {@code flink-s3-fs-presto} and with AWS SDK v2 in {@code
+ * flink-s3-fs-hadoop}.
+ */
@Internal
public abstract class AbstractS3DelegationTokenProvider implements DelegationTokenProvider {
@@ -51,24 +52,21 @@ public abstract class AbstractS3DelegationTokenProvider implements DelegationTok
public void init(Configuration configuration) {
region = configuration.getString(String.format("%s.region", serviceConfigPrefix()), null);
if (!StringUtils.isNullOrWhitespaceOnly(region)) {
- LOG.debug("Region: " + region);
+ LOG.debug("Region: {}", region);
}
accessKey =
configuration.getString(
String.format("%s.access-key", serviceConfigPrefix()), null);
if (!StringUtils.isNullOrWhitespaceOnly(accessKey)) {
- LOG.debug("Access key: " + accessKey);
+ LOG.debug("Access key: {}", accessKey);
}
secretKey =
configuration.getString(
String.format("%s.secret-key", serviceConfigPrefix()), null);
if (!StringUtils.isNullOrWhitespaceOnly(secretKey)) {
- LOG.debug(
- "Secret key: "
- + GlobalConfiguration.HIDDEN_CONTENT
- + " (sensitive information)");
+ LOG.debug("Secret key: {} (sensitive information)", GlobalConfiguration.HIDDEN_CONTENT);
}
}
@@ -87,22 +85,21 @@ public boolean delegationTokensRequired() {
public ObtainedDelegationTokens obtainDelegationTokens() throws Exception {
LOG.info("Obtaining session credentials token with access key: {}", accessKey);
- AWSSecurityTokenService stsClient =
- AWSSecurityTokenServiceClientBuilder.standard()
- .withRegion(region)
- .withCredentials(
- new AWSStaticCredentialsProvider(
- new BasicAWSCredentials(accessKey, secretKey)))
- .build();
- GetSessionTokenResult sessionTokenResult = stsClient.getSessionToken();
- Credentials credentials = sessionTokenResult.getCredentials();
+ S3SessionCredentials credentials = getSessionCredentials(region, accessKey, secretKey);
LOG.info(
"Session credentials obtained successfully with access key: {} expiration: {}",
credentials.getAccessKeyId(),
- credentials.getExpiration());
+ credentials.getExpirationEpochMilli());
return new ObtainedDelegationTokens(
InstantiationUtil.serializeObject(credentials),
- Optional.of(credentials.getExpiration().getTime()));
+ Optional.of(credentials.getExpirationEpochMilli()));
}
+
+ /**
+ * Obtains session credentials from AWS STS with the SDK bundled into the concrete filesystem
+ * plugin.
+ */
+ protected abstract S3SessionCredentials getSessionCredentials(
+ String region, String accessKey, String secretKey) throws Exception;
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiver.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiver.java
index bad552f432b68e..420a1b5a452260 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiver.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiver.java
@@ -26,13 +26,23 @@
import org.apache.flink.util.InstantiationUtil;
import org.apache.flink.util.StringUtils;
-import com.amazonaws.services.securitytoken.model.Credentials;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
-/** Delegation token receiver for S3 filesystems. */
+import java.util.LinkedHashSet;
+import java.util.Set;
+
+/**
+ * Delegation token receiver for S3 filesystems.
+ *
+ *
This class deliberately references no AWS SDK types (see {@link S3SessionCredentials}), so it
+ * can be loaded both in the {@code flink-s3-fs-presto} plugin (AWS SDK v1 only) and the {@code
+ * flink-s3-fs-hadoop} plugin (AWS SDK v2 only). The received credentials are exposed through {@link
+ * #getCredentials()} to the SDK-specific credential providers and to SDK-agnostic consumers such as
+ * the s5cmd integration in {@code FlinkS3FileSystem}.
+ */
@Internal
public abstract class AbstractS3DelegationTokenReceiver implements DelegationTokenReceiver {
@@ -41,20 +51,45 @@ public abstract class AbstractS3DelegationTokenReceiver implements DelegationTok
private static final Logger LOG =
LoggerFactory.getLogger(AbstractS3DelegationTokenReceiver.class);
- @VisibleForTesting @Nullable static volatile Credentials credentials;
+ @VisibleForTesting @Nullable static volatile S3SessionCredentials credentials;
@VisibleForTesting @Nullable static volatile String region;
public static void updateHadoopConfig(org.apache.hadoop.conf.Configuration hadoopConfig) {
+ updateHadoopConfig(hadoopConfig, DynamicTemporaryAWSCredentialsProvider.NAME);
+ }
+
+ /**
+ * Registers the given delegation token credentials provider in {@code
+ * fs.s3a.aws.credentials.provider} and propagates the configured region.
+ *
+ *
When {@code credentialsProviderName} differs from the SDK v1 based {@link
+ * DynamicTemporaryAWSCredentialsProvider}, any user-configured reference to that legacy
+ * provider is remapped to {@code credentialsProviderName}: the legacy provider cannot be loaded
+ * in plugins that do not bundle AWS SDK v1 (e.g. {@code flink-s3-fs-hadoop}).
+ */
+ public static void updateHadoopConfig(
+ org.apache.hadoop.conf.Configuration hadoopConfig, String credentialsProviderName) {
LOG.info("Updating Hadoop configuration");
String providers = hadoopConfig.get(PROVIDER_CONFIG_NAME, "");
- if (!providers.contains(DynamicTemporaryAWSCredentialsProvider.NAME)) {
+ if (!credentialsProviderName.equals(DynamicTemporaryAWSCredentialsProvider.NAME)) {
+ String remappedProviders = replaceLegacyProvider(providers, credentialsProviderName);
+ if (!remappedProviders.equals(providers)) {
+ LOG.info(
+ "Remapped legacy SDK v1 credentials provider {} to {}",
+ DynamicTemporaryAWSCredentialsProvider.NAME,
+ credentialsProviderName);
+ providers = remappedProviders;
+ hadoopConfig.set(PROVIDER_CONFIG_NAME, providers);
+ }
+ }
+ if (!providers.contains(credentialsProviderName)) {
if (providers.isEmpty()) {
LOG.debug("Setting provider");
- providers = DynamicTemporaryAWSCredentialsProvider.NAME;
+ providers = credentialsProviderName;
} else {
- providers = DynamicTemporaryAWSCredentialsProvider.NAME + "," + providers;
+ providers = credentialsProviderName + "," + providers;
LOG.debug("Prepending provider, new providers value: {}", providers);
}
hadoopConfig.set(PROVIDER_CONFIG_NAME, providers);
@@ -70,6 +105,28 @@ public static void updateHadoopConfig(org.apache.hadoop.conf.Configuration hadoo
LOG.info("Updated Hadoop configuration successfully");
}
+ /**
+ * Replaces the legacy SDK v1 delegation token credentials provider with its replacement,
+ * dropping duplicates while preserving the order of the remaining chain. Returns the input
+ * unchanged when the legacy provider is not referenced.
+ */
+ private static String replaceLegacyProvider(String providers, String replacement) {
+ boolean legacyFound = false;
+ Set remapped = new LinkedHashSet<>();
+ for (String provider : providers.split(",")) {
+ String trimmed = provider.trim();
+ if (trimmed.isEmpty()) {
+ continue;
+ }
+ if (trimmed.equals(DynamicTemporaryAWSCredentialsProvider.NAME)) {
+ legacyFound = true;
+ trimmed = replacement;
+ }
+ remapped.add(trimmed);
+ }
+ return legacyFound ? String.join(",", remapped) : providers;
+ }
+
@Override
public void init(Configuration configuration) {
region =
@@ -79,24 +136,25 @@ public void init(Configuration configuration) {
DelegationTokenProvider.CONFIG_PREFIX, serviceName()),
null);
if (!StringUtils.isNullOrWhitespaceOnly(region)) {
- LOG.debug("Region: " + region);
+ LOG.debug("Region: {}", region);
}
}
@Override
public void onNewTokensObtained(byte[] tokens) throws Exception {
LOG.info("Updating session credentials");
- credentials =
+ S3SessionCredentials newCredentials =
InstantiationUtil.deserializeObject(
tokens, AbstractS3DelegationTokenReceiver.class.getClassLoader());
+ credentials = newCredentials;
LOG.info(
"Session credentials updated successfully with access key: {} expiration: {}",
- credentials.getAccessKeyId(),
- credentials.getExpiration());
+ newCredentials.getAccessKeyId(),
+ newCredentials.getExpirationEpochMilli());
}
@Nullable
- public static Credentials getCredentials() {
+ public static S3SessionCredentials getCredentials() {
return credentials;
}
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProvider.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProvider.java
index 36398448ca38ce..0d9d1522d6afb4 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProvider.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProvider.java
@@ -21,12 +21,11 @@
import org.apache.flink.annotation.Internal;
import com.amazonaws.SdkBaseException;
+import com.amazonaws.SdkClientException;
import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.BasicSessionCredentials;
-import com.amazonaws.services.securitytoken.model.Credentials;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -35,13 +34,26 @@
/**
* Support dynamic session credentials for authenticating with AWS. Please note that users may
* reference this class name from configuration property fs.s3a.aws.credentials.provider. Therefore,
- * changing the class name would be a backward-incompatible change. This credential provider must
- * not fail in creation because that will break a chain of credential providers.
+ * changing the class name would be a backward-incompatible change. This class is based on AWS SDK
+ * v1 and serves the flink-s3-fs-presto plugin; the SDK v2 based flink-s3-fs-hadoop plugin cannot
+ * load it and remaps this class name to {@code HadoopDynamicTemporaryAWSCredentialsProvider} in its
+ * Hadoop configuration. This credential provider must not fail in creation because that will break
+ * a chain of credential providers. When no credentials are available yet, {@link #getCredentials()}
+ * throws the SDK v1 {@link SdkClientException} (rather than Hadoop's s3a {@code
+ * NoAwsCredentialsException}, whose class hierarchy is based on AWS SDK v2 since Hadoop 3.4 and
+ * therefore cannot be loaded inside the flink-s3-fs-presto plugin, which bundles only SDK v1);
+ * credential provider chains treat any such exception as a signal to move on to the next provider.
*/
@Internal
public class DynamicTemporaryAWSCredentialsProvider implements AWSCredentialsProvider {
- public static final String NAME = DynamicTemporaryAWSCredentialsProvider.class.getName();
+ /**
+ * Spelled out as a string literal (a compile-time constant) so that referencing {@code NAME}
+ * never triggers loading this class: it implements an SDK v1 interface that is absent from the
+ * flink-s3-fs-hadoop jar. Pinned to the actual class name by a test.
+ */
+ public static final String NAME =
+ "org.apache.flink.fs.s3.common.token.DynamicTemporaryAWSCredentialsProvider";
public static final String COMPONENT = "Dynamic session credentials for Flink";
@@ -54,9 +66,9 @@ public DynamicTemporaryAWSCredentialsProvider(URI uri, Configuration conf) {}
@Override
public AWSCredentials getCredentials() throws SdkBaseException {
- Credentials credentials = AbstractS3DelegationTokenReceiver.getCredentials();
+ S3SessionCredentials credentials = AbstractS3DelegationTokenReceiver.getCredentials();
if (credentials == null) {
- throw new NoAwsCredentialsException(COMPONENT);
+ throw new SdkClientException(COMPONENT + ": No AWS credentials");
}
LOG.debug("Providing session credentials");
return new BasicSessionCredentials(
@@ -67,6 +79,6 @@ public AWSCredentials getCredentials() throws SdkBaseException {
@Override
public void refresh() {
- // Intentionally blank. Credentials are updated by S3DelegationTokenReceiver
+ // Intentionally blank. Credentials are updated by the delegation token receiver.
}
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/S3SessionCredentials.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/S3SessionCredentials.java
new file mode 100644
index 00000000000000..ac5bcb77e6a3d3
--- /dev/null
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/token/S3SessionCredentials.java
@@ -0,0 +1,80 @@
+/*
+ * 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.flink.fs.s3.common.token;
+
+import org.apache.flink.annotation.Internal;
+
+import javax.annotation.Nullable;
+
+import java.io.Serializable;
+
+/**
+ * AWS session credentials obtained through delegation tokens, held in an SDK-agnostic form.
+ *
+ * This class deliberately references no AWS SDK types: it is bundled into both the {@code
+ * flink-s3-fs-hadoop} jar (which ships only AWS SDK v2) and the {@code flink-s3-fs-presto} jar
+ * (which ships only AWS SDK v1), so it must be loadable with either SDK absent. Each filesystem's
+ * credential provider converts these values into its own SDK's credential type.
+ *
+ *
Instances are Java-serialized as the delegation token payload sent from the JobManager-side
+ * {@link AbstractS3DelegationTokenProvider} to the TaskManager-side {@link
+ * AbstractS3DelegationTokenReceiver}. Both endpoints are always loaded from the same plugin jar, so
+ * the payload never crosses versions and is never persisted.
+ */
+@Internal
+public final class S3SessionCredentials implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final String accessKeyId;
+
+ private final String secretAccessKey;
+
+ @Nullable private final String sessionToken;
+
+ private final long expirationEpochMilli;
+
+ public S3SessionCredentials(
+ String accessKeyId,
+ String secretAccessKey,
+ @Nullable String sessionToken,
+ long expirationEpochMilli) {
+ this.accessKeyId = accessKeyId;
+ this.secretAccessKey = secretAccessKey;
+ this.sessionToken = sessionToken;
+ this.expirationEpochMilli = expirationEpochMilli;
+ }
+
+ public String getAccessKeyId() {
+ return accessKeyId;
+ }
+
+ public String getSecretAccessKey() {
+ return secretAccessKey;
+ }
+
+ @Nullable
+ public String getSessionToken() {
+ return sessionToken;
+ }
+
+ public long getExpirationEpochMilli() {
+ return expirationEpochMilli;
+ }
+}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/MultiPartUploadInfo.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/MultiPartUploadInfo.java
index d76d2889512753..b4e9883be1339e 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/MultiPartUploadInfo.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/MultiPartUploadInfo.java
@@ -20,7 +20,7 @@
import org.apache.flink.annotation.Internal;
-import com.amazonaws.services.s3.model.PartETag;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
import java.io.File;
import java.util.ArrayList;
@@ -38,7 +38,7 @@ final class MultiPartUploadInfo {
private final String uploadId;
- private final List completeParts;
+ private final List completeParts;
private final Optional incompletePart;
@@ -57,7 +57,7 @@ final class MultiPartUploadInfo {
MultiPartUploadInfo(
final String objectName,
final String uploadId,
- final List completeParts,
+ final List completeParts,
final long numBytes,
final Optional incompletePart) {
@@ -92,7 +92,7 @@ Optional getIncompletePart() {
return incompletePart;
}
- List getCopyOfEtagsOfCompleteParts() {
+ List getCopyOfEtagsOfCompleteParts() {
return new ArrayList<>(completeParts);
}
@@ -101,7 +101,7 @@ void registerNewPart(long length) {
this.numberOfRegisteredParts++;
}
- void registerCompletePart(PartETag eTag) {
+ void registerCompletePart(CompletedPart eTag) {
completeParts.add(eTag);
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImpl.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImpl.java
index ddc6ceb83a6b56..4f53098419a939 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImpl.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImpl.java
@@ -22,8 +22,8 @@
import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.core.fs.RefCountedFSOutputStream;
-import com.amazonaws.services.s3.model.PartETag;
-import com.amazonaws.services.s3.model.UploadPartResult;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
import javax.annotation.Nullable;
import javax.annotation.concurrent.NotThreadSafe;
@@ -61,7 +61,7 @@ final class RecoverableMultiPartUploadImpl implements RecoverableMultiPartUpload
private final Executor uploadThreadPool;
- private final Deque> uploadsInProgress;
+ private final Deque> uploadsInProgress;
private final String namePrefixForTempObjects;
@@ -74,7 +74,7 @@ private RecoverableMultiPartUploadImpl(
Executor uploadThreadPool,
String uploadId,
String objectName,
- List partsSoFar,
+ List partsSoFar,
long numBytes,
Optional incompletePart) {
checkArgument(numBytes >= 0L);
@@ -102,7 +102,7 @@ public void uploadPart(RefCountedFSOutputStream file) throws IOException {
// writing to the file we are uploading.
checkState(file.isClosed());
- final CompletableFuture future = new CompletableFuture<>();
+ final CompletableFuture future = new CompletableFuture<>();
uploadsInProgress.add(future);
final long partLength = file.getPos();
@@ -153,7 +153,8 @@ public S3Recoverable snapshotAndGetRecoverable(
final String objectName = currentUploadInfo.getObjectName();
final String uploadId = currentUploadInfo.getUploadId();
- final List completedParts = currentUploadInfo.getCopyOfEtagsOfCompleteParts();
+ final List completedParts =
+ currentUploadInfo.getCopyOfEtagsOfCompleteParts();
final long sizeInBytes = currentUploadInfo.getExpectedSizeInBytes();
if (incompletePartObjectName == null) {
@@ -193,6 +194,25 @@ private String safelyUploadSmallPart(@Nullable RefCountedFSOutputStream file)
// utils
// ------------------------------------------------------------------------
+ /**
+ * Builds the {@link CompletedPart} for the CompleteMultipartUpload request from an upload-part
+ * response. Besides the ETag, every per-part checksum is copied: when checksum generation is
+ * enabled ({@code fs.s3a.checksum.generation}), S3 rejects a CompleteMultipartUpload whose
+ * parts do not repeat the checksum they were uploaded with.
+ */
+ @VisibleForTesting
+ static CompletedPart completedPartFromResponse(int partNumber, UploadPartResponse response) {
+ final CompletedPart.Builder builder =
+ CompletedPart.builder().partNumber(partNumber).eTag(response.eTag());
+ for (S3PartChecksum checksum : S3PartChecksum.values()) {
+ final String value = checksum.valueOf(response);
+ if (value != null) {
+ checksum.applyTo(builder, value);
+ }
+ }
+ return builder.build();
+ }
+
@VisibleForTesting
static String createIncompletePartObjectNamePrefix(String objectName) {
checkNotNull(objectName);
@@ -219,16 +239,16 @@ private void awaitPendingPartsUpload() throws IOException {
checkState(currentUploadInfo.getRemainingParts() == uploadsInProgress.size());
while (currentUploadInfo.getRemainingParts() > 0) {
- CompletableFuture next = uploadsInProgress.peekFirst();
- PartETag nextPart = awaitPendingPartUploadToComplete(next);
+ CompletableFuture next = uploadsInProgress.peekFirst();
+ CompletedPart nextPart = awaitPendingPartUploadToComplete(next);
currentUploadInfo.registerCompletePart(nextPart);
uploadsInProgress.removeFirst();
}
}
- private PartETag awaitPendingPartUploadToComplete(CompletableFuture upload)
+ private CompletedPart awaitPendingPartUploadToComplete(CompletableFuture upload)
throws IOException {
- final PartETag completedUploadEtag;
+ final CompletedPart completedUploadEtag;
try {
completedUploadEtag = upload.get();
} catch (InterruptedException e) {
@@ -267,7 +287,7 @@ public static RecoverableMultiPartUploadImpl recoverUpload(
final Executor uploadThreadPool,
final String multipartUploadId,
final String objectName,
- final List partsSoFar,
+ final List partsSoFar,
final long numBytesSoFar,
final Optional incompletePart) {
@@ -297,13 +317,13 @@ private static class UploadTask implements Runnable {
private final RefCountedFSOutputStream file;
- private final CompletableFuture future;
+ private final CompletableFuture future;
UploadTask(
final S3AccessHelper s3AccessHelper,
final MultiPartUploadInfo currentUpload,
final RefCountedFSOutputStream file,
- final CompletableFuture future) {
+ final CompletableFuture future) {
checkNotNull(currentUpload);
@@ -322,14 +342,14 @@ private static class UploadTask implements Runnable {
@Override
public void run() {
try {
- final UploadPartResult result =
+ final UploadPartResponse result =
s3AccessHelper.uploadPart(
objectName,
uploadId,
partNumber,
file.getInputFile(),
file.getPos());
- future.complete(new PartETag(result.getPartNumber(), result.getETag()));
+ future.complete(completedPartFromResponse(partNumber, result));
file.release();
} catch (Throwable t) {
future.completeExceptionally(t);
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3AccessHelper.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3AccessHelper.java
index 13f73ee4c9818d..31e1060a59c18f 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3AccessHelper.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3AccessHelper.java
@@ -20,11 +20,11 @@
import org.apache.flink.annotation.Internal;
-import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
-import com.amazonaws.services.s3.model.ObjectMetadata;
-import com.amazonaws.services.s3.model.PartETag;
-import com.amazonaws.services.s3.model.PutObjectResult;
-import com.amazonaws.services.s3.model.UploadPartResult;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
import java.io.File;
import java.io.IOException;
@@ -33,12 +33,15 @@
/**
* An interface that abstracts away the Multi-Part Upload (MPU) functionality offered by S3, from
- * the specific implementation of the file system. This is needed so that we can accommodate both
- * Hadoop S3 and Presto.
+ * the specific implementation of the file system.
*
- * Multipart uploads are convenient for large object. These will be uploaded in multiple parts
- * and the mutli-part upload is the equivalent of a transaction, where the upload with all its parts
+ *
Multipart uploads are convenient for large objects. These will be uploaded in multiple parts
+ * and the multipart upload is the equivalent of a transaction, where the upload with all its parts
* will be either committed or discarded.
+ *
+ *
This interface uses AWS SDK v2 types and is implemented by the Hadoop S3A filesystem module.
+ * The Presto S3 filesystem (which uses AWS SDK v1) returns null for {@code S3AccessHelper} and
+ * handles uploads through its own mechanism.
*/
@Internal
public interface S3AccessHelper {
@@ -61,10 +64,10 @@ public interface S3AccessHelper {
* @param partNumber the number of the part being uploaded (has to be in [1 ... 10000]).
* @param inputFile the (local) file holding the part to be uploaded.
* @param length the length of the part.
- * @return The {@link UploadPartResult result} of the attempt to upload the part.
+ * @return The {@link UploadPartResponse result} of the attempt to upload the part.
* @throws IOException
*/
- UploadPartResult uploadPart(
+ UploadPartResponse uploadPart(
String key, String uploadId, int partNumber, File inputFile, long length)
throws IOException;
@@ -75,26 +78,27 @@ UploadPartResult uploadPart(
*
* @param key the key used to identify this part.
* @param inputFile the (local) file holding the data to be uploaded.
- * @return The {@link PutObjectResult result} of the attempt to stage the incomplete part.
+ * @return The {@link PutObjectResponse result} of the attempt to stage the incomplete part.
* @throws IOException
*/
- PutObjectResult putObject(String key, File inputFile) throws IOException;
+ PutObjectResponse putObject(String key, File inputFile) throws IOException;
/**
* Finalizes a Multi-Part Upload.
*
* @param key the key identifying the object we finished uploading.
* @param uploadId the id of the MPU.
- * @param partETags the list of {@link PartETag ETags} associated with this MPU.
+ * @param partETags the list of {@link CompletedPart ETags} associated with this MPU.
* @param length the size of the uploaded object.
* @param errorCount a counter that will be used to count any failed attempts to commit the MPU.
- * @return The {@link CompleteMultipartUploadResult result} of the attempt to finalize the MPU.
+ * @return The {@link CompleteMultipartUploadResponse result} of the attempt to finalize the
+ * MPU.
* @throws IOException
*/
- CompleteMultipartUploadResult commitMultiPartUpload(
+ CompleteMultipartUploadResponse commitMultiPartUpload(
String key,
String uploadId,
- List partETags,
+ List partETags,
long length,
AtomicInteger errorCount)
throws IOException;
@@ -124,8 +128,8 @@ CompleteMultipartUploadResult commitMultiPartUpload(
* Fetches the metadata associated with a given key on S3.
*
* @param key the key.
- * @return The associated {@link ObjectMetadata}.
+ * @return The associated {@link HeadObjectResponse}.
* @throws IOException
*/
- ObjectMetadata getObjectMetadata(String key) throws IOException;
+ HeadObjectResponse getObjectMetadata(String key) throws IOException;
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Committer.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Committer.java
index 6cea633b30b05c..891b7644e70a59 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Committer.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Committer.java
@@ -21,10 +21,10 @@
import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
-import com.amazonaws.services.s3.model.ObjectMetadata;
-import com.amazonaws.services.s3.model.PartETag;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
import java.io.FileNotFoundException;
import java.io.IOException;
@@ -44,7 +44,7 @@ public final class S3Committer implements RecoverableFsDataOutputStream.Committe
private final String objectName;
- private final List parts;
+ private final List parts;
private final long totalLength;
@@ -52,7 +52,7 @@ public final class S3Committer implements RecoverableFsDataOutputStream.Committe
S3AccessHelper s3AccessHelper,
String objectName,
String uploadId,
- List parts,
+ List parts,
long totalLength) {
this.s3AccessHelper = checkNotNull(s3AccessHelper);
this.objectName = checkNotNull(objectName);
@@ -101,8 +101,8 @@ public void commitAfterRecovery() throws IOException {
LOG.trace("Exception when committing:", e);
try {
- ObjectMetadata metadata = s3AccessHelper.getObjectMetadata(objectName);
- if (totalLength != metadata.getContentLength()) {
+ HeadObjectResponse metadata = s3AccessHelper.getObjectMetadata(objectName);
+ if (totalLength != metadata.contentLength()) {
String message =
String.format(
"Inconsistent result for object %s: conflicting lengths. "
@@ -110,7 +110,7 @@ public void commitAfterRecovery() throws IOException {
objectName,
uploadId,
totalLength,
- metadata.getContentLength());
+ metadata.contentLength());
LOG.warn(message);
throw new IOException(message, e);
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3PartChecksum.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3PartChecksum.java
new file mode 100644
index 00000000000000..8d17db03f47988
--- /dev/null
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3PartChecksum.java
@@ -0,0 +1,140 @@
+/*
+ * 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.flink.fs.s3.common.writer;
+
+import org.apache.flink.annotation.Internal;
+
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+/**
+ * The per-part checksums an S3 {@link UploadPartResponse} can carry, together with the stable wire
+ * tags used by {@link S3RecoverableSerializer}.
+ *
+ * This enum is the single source of truth both for propagating checksums from {@link
+ * UploadPartResponse} to {@link CompletedPart} and for (de)serializing them in recoverable state,
+ * so the two sets can never drift apart. Adding a checksum field introduced by a newer AWS SDK only
+ * requires a new constant. Wire tags are persisted in checkpointed state and must never be
+ * renumbered or reused.
+ */
+@Internal
+enum S3PartChecksum {
+ CRC32(
+ (byte) 1,
+ UploadPartResponse::checksumCRC32,
+ CompletedPart::checksumCRC32,
+ CompletedPart.Builder::checksumCRC32),
+ CRC32C(
+ (byte) 2,
+ UploadPartResponse::checksumCRC32C,
+ CompletedPart::checksumCRC32C,
+ CompletedPart.Builder::checksumCRC32C),
+ CRC64NVME(
+ (byte) 3,
+ UploadPartResponse::checksumCRC64NVME,
+ CompletedPart::checksumCRC64NVME,
+ CompletedPart.Builder::checksumCRC64NVME),
+ SHA1(
+ (byte) 4,
+ UploadPartResponse::checksumSHA1,
+ CompletedPart::checksumSHA1,
+ CompletedPart.Builder::checksumSHA1),
+ SHA256(
+ (byte) 5,
+ UploadPartResponse::checksumSHA256,
+ CompletedPart::checksumSHA256,
+ CompletedPart.Builder::checksumSHA256),
+ SHA512(
+ (byte) 6,
+ UploadPartResponse::checksumSHA512,
+ CompletedPart::checksumSHA512,
+ CompletedPart.Builder::checksumSHA512),
+ MD5(
+ (byte) 7,
+ UploadPartResponse::checksumMD5,
+ CompletedPart::checksumMD5,
+ CompletedPart.Builder::checksumMD5),
+ XXHASH64(
+ (byte) 8,
+ UploadPartResponse::checksumXXHASH64,
+ CompletedPart::checksumXXHASH64,
+ CompletedPart.Builder::checksumXXHASH64),
+ XXHASH3(
+ (byte) 9,
+ UploadPartResponse::checksumXXHASH3,
+ CompletedPart::checksumXXHASH3,
+ CompletedPart.Builder::checksumXXHASH3),
+ XXHASH128(
+ (byte) 10,
+ UploadPartResponse::checksumXXHASH128,
+ CompletedPart::checksumXXHASH128,
+ CompletedPart.Builder::checksumXXHASH128);
+
+ private final byte wireTag;
+
+ private final Function responseGetter;
+
+ private final Function partGetter;
+
+ private final BiConsumer builderSetter;
+
+ S3PartChecksum(
+ byte wireTag,
+ Function responseGetter,
+ Function partGetter,
+ BiConsumer builderSetter) {
+ this.wireTag = wireTag;
+ this.responseGetter = responseGetter;
+ this.partGetter = partGetter;
+ this.builderSetter = builderSetter;
+ }
+
+ byte getWireTag() {
+ return wireTag;
+ }
+
+ @Nullable
+ String valueOf(UploadPartResponse response) {
+ return responseGetter.apply(response);
+ }
+
+ @Nullable
+ String valueOf(CompletedPart part) {
+ return partGetter.apply(part);
+ }
+
+ void applyTo(CompletedPart.Builder builder, String value) {
+ builderSetter.accept(builder, value);
+ }
+
+ static S3PartChecksum fromWireTag(byte wireTag) throws IOException {
+ for (S3PartChecksum checksum : values()) {
+ if (checksum.wireTag == wireTag) {
+ return checksum;
+ }
+ }
+ throw new IOException("Corrupt data: Unknown part checksum tag " + wireTag);
+ }
+}
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Recoverable.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Recoverable.java
index cbe9e159b07fc6..d4757626529e85 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Recoverable.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3Recoverable.java
@@ -20,7 +20,7 @@
import org.apache.flink.core.fs.RecoverableWriter;
-import com.amazonaws.services.s3.model.PartETag;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
import javax.annotation.Nullable;
@@ -36,7 +36,7 @@ public final class S3Recoverable implements RecoverableWriter.ResumeRecoverable
private final String objectName;
- private final List parts;
+ private final List parts;
@Nullable private final String lastPartObject;
@@ -44,14 +44,15 @@ public final class S3Recoverable implements RecoverableWriter.ResumeRecoverable
private long lastPartObjectLength;
- S3Recoverable(String objectName, String uploadId, List parts, long numBytesInParts) {
+ S3Recoverable(
+ String objectName, String uploadId, List parts, long numBytesInParts) {
this(objectName, uploadId, parts, numBytesInParts, null, -1L);
}
S3Recoverable(
String objectName,
String uploadId,
- List parts,
+ List parts,
long numBytesInParts,
@Nullable String lastPartObject,
long lastPartObjectLength) {
@@ -77,7 +78,7 @@ public String getObjectName() {
return objectName;
}
- public List parts() {
+ public List parts() {
return parts;
}
@@ -105,11 +106,11 @@ public String toString() {
buf.append(", bytesInParts=").append(numBytesInParts);
buf.append(", parts=[");
int num = 0;
- for (PartETag part : parts) {
+ for (CompletedPart part : parts) {
if (0 != num++) {
buf.append(", ");
}
- buf.append(part.getPartNumber()).append('=').append(part.getETag());
+ buf.append(part.partNumber()).append('=').append(part.eTag());
}
buf.append("], trailingPart=").append(lastPartObject);
buf.append("trailingPartLen=").append(lastPartObjectLength);
diff --git a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
index a72e99be831d13..c5dfa13873abb0 100644
--- a/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
+++ b/flink-filesystems/flink-s3-fs-base/src/main/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializer.java
@@ -21,7 +21,7 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.core.io.SimpleVersionedSerializer;
-import com.amazonaws.services.s3.model.PartETag;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
import java.io.IOException;
import java.nio.ByteBuffer;
@@ -31,7 +31,14 @@
import java.util.ArrayList;
import java.util.List;
-/** Serializer implementation for a {@link S3Recoverable}. */
+/**
+ * Serializer implementation for a {@link S3Recoverable}.
+ *
+ * Version 2 extends version 1 by persisting the per-part checksums (see {@link S3PartChecksum})
+ * so that a CompleteMultipartUpload after recovery can repeat the checksum each part was uploaded
+ * with. Version-1 state deserializes into checksum-less parts, which is correct: it was written by
+ * releases that never generated part checksums.
+ */
@Internal
final class S3RecoverableSerializer implements SimpleVersionedSerializer {
@@ -46,22 +53,24 @@ private S3RecoverableSerializer() {}
@Override
public int getVersion() {
- return 1;
+ return 2;
}
@Override
public byte[] serialize(S3Recoverable obj) throws IOException {
- final List partList = obj.parts();
- final PartETag[] parts = partList.toArray(new PartETag[partList.size()]);
+ final List partList = obj.parts();
+ final CompletedPart[] parts = partList.toArray(new CompletedPart[0]);
final byte[] keyBytes = obj.getObjectName().getBytes(CHARSET);
final byte[] uploadIdBytes = obj.uploadId().getBytes(CHARSET);
final byte[][] etags = new byte[parts.length][];
+ final byte[][] checksumBlocks = new byte[parts.length][];
int partEtagBytes = 0;
for (int i = 0; i < parts.length; i++) {
- etags[i] = parts[i].getETag().getBytes(CHARSET);
- partEtagBytes += etags[i].length + 2 * Integer.BYTES;
+ etags[i] = parts[i].eTag().getBytes(CHARSET);
+ checksumBlocks[i] = encodeChecksums(parts[i]);
+ partEtagBytes += etags[i].length + 2 * Integer.BYTES + checksumBlocks[i].length;
}
final String lastObjectKey = obj.incompleteObjectName();
@@ -93,10 +102,11 @@ public byte[] serialize(S3Recoverable obj) throws IOException {
bb.putInt(etags.length);
for (int i = 0; i < parts.length; i++) {
- PartETag pe = parts[i];
- bb.putInt(pe.getPartNumber());
+ CompletedPart pe = parts[i];
+ bb.putInt(pe.partNumber());
bb.putInt(etags[i].length);
bb.put(etags[i]);
+ bb.put(checksumBlocks[i]);
}
bb.putLong(obj.numBytesInParts());
@@ -113,11 +123,41 @@ public byte[] serialize(S3Recoverable obj) throws IOException {
return targetBytes;
}
+ /**
+ * Encodes the checksums of a part as: checksum count (byte, 0 for none), then per checksum its
+ * wire tag (byte) and the length-prefixed UTF-8 value exactly as the SDK returned it.
+ */
+ private static byte[] encodeChecksums(CompletedPart part) {
+ final List presentChecksums = new ArrayList<>();
+ final List values = new ArrayList<>();
+ int size = Byte.BYTES;
+ for (S3PartChecksum checksum : S3PartChecksum.values()) {
+ final String value = checksum.valueOf(part);
+ if (value != null) {
+ final byte[] valueBytes = value.getBytes(CHARSET);
+ presentChecksums.add(checksum);
+ values.add(valueBytes);
+ size += Byte.BYTES + Integer.BYTES + valueBytes.length;
+ }
+ }
+
+ final ByteBuffer bb = ByteBuffer.allocate(size).order(ByteOrder.LITTLE_ENDIAN);
+ bb.put((byte) presentChecksums.size());
+ for (int i = 0; i < presentChecksums.size(); i++) {
+ bb.put(presentChecksums.get(i).getWireTag());
+ bb.putInt(values.get(i).length);
+ bb.put(values.get(i));
+ }
+ return bb.array();
+ }
+
@Override
public S3Recoverable deserialize(int version, byte[] serialized) throws IOException {
switch (version) {
case 1:
return deserializeV1(serialized);
+ case 2:
+ return deserializeV2(serialized);
default:
throw new IOException("Unrecognized version or corrupt state: " + version);
}
@@ -137,14 +177,58 @@ private static S3Recoverable deserializeV1(byte[] serialized) throws IOException
bb.get(uploadIdBytes);
final int numParts = bb.getInt();
- final ArrayList parts = new ArrayList<>(numParts);
+ final ArrayList parts = new ArrayList<>(numParts);
+ for (int i = 0; i < numParts; i++) {
+ final int partNum = bb.getInt();
+ final byte[] buffer = new byte[bb.getInt()];
+ bb.get(buffer);
+ parts.add(
+ CompletedPart.builder()
+ .partNumber(partNum)
+ .eTag(new String(buffer, CHARSET))
+ .build());
+ }
+
+ return deserializeTrailer(bb, keyBytes, uploadIdBytes, parts);
+ }
+
+ private static S3Recoverable deserializeV2(byte[] serialized) throws IOException {
+ final ByteBuffer bb = ByteBuffer.wrap(serialized).order(ByteOrder.LITTLE_ENDIAN);
+
+ if (bb.getInt() != MAGIC_NUMBER) {
+ throw new IOException("Corrupt data: Unexpected magic number.");
+ }
+
+ final byte[] keyBytes = new byte[bb.getInt()];
+ bb.get(keyBytes);
+
+ final byte[] uploadIdBytes = new byte[bb.getInt()];
+ bb.get(uploadIdBytes);
+
+ final int numParts = bb.getInt();
+ final ArrayList parts = new ArrayList<>(numParts);
for (int i = 0; i < numParts; i++) {
final int partNum = bb.getInt();
final byte[] buffer = new byte[bb.getInt()];
bb.get(buffer);
- parts.add(new PartETag(partNum, new String(buffer, CHARSET)));
+ final CompletedPart.Builder partBuilder =
+ CompletedPart.builder().partNumber(partNum).eTag(new String(buffer, CHARSET));
+ final int numChecksums = bb.get();
+ for (int c = 0; c < numChecksums; c++) {
+ final S3PartChecksum checksum = S3PartChecksum.fromWireTag(bb.get());
+ final byte[] valueBuffer = new byte[bb.getInt()];
+ bb.get(valueBuffer);
+ checksum.applyTo(partBuilder, new String(valueBuffer, CHARSET));
+ }
+ parts.add(partBuilder.build());
}
+ return deserializeTrailer(bb, keyBytes, uploadIdBytes, parts);
+ }
+
+ /** Reads the fields following the part list; identical in versions 1 and 2. */
+ private static S3Recoverable deserializeTrailer(
+ ByteBuffer bb, byte[] keyBytes, byte[] uploadIdBytes, List parts) {
final long numBytes = bb.getLong();
final String lastPart;
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAApplicationRunOnSeaweedFsS3StoreITCase.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAApplicationRunOnSeaweedFsS3StoreITCase.java
index cb5a7521f430da..2827ee1b5d5f16 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAApplicationRunOnSeaweedFsS3StoreITCase.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAApplicationRunOnSeaweedFsS3StoreITCase.java
@@ -100,6 +100,12 @@ private static Configuration createConfiguration() {
getSeaweedFsContainer().setS3ConfigOptions(config);
+ // S3A on Hadoop 3.4 keeps zero-byte directory markers by default
+ // (fs.s3a.directory.marker.retention=keep); pin the pre-3.4 "delete" policy so the raw
+ // listings below contain only actual result-store entries. The Presto subclass maps this
+ // key to presto.s3.directory.marker.retention, which PrestoS3FileSystem ignores.
+ config.setString("s3.directory.marker.retention", "delete");
+
// ApplicationResultStore configuration
config.set(ApplicationResultStoreOptions.DELETE_ON_COMMIT, Boolean.FALSE);
config.set(
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAJobRunOnSeaweedFsS3StoreITCase.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAJobRunOnSeaweedFsS3StoreITCase.java
index d047d36f2d0a6b..733daff0321fba 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAJobRunOnSeaweedFsS3StoreITCase.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/HAJobRunOnSeaweedFsS3StoreITCase.java
@@ -99,6 +99,12 @@ private static Configuration createConfiguration() {
getSeaweedFsContainer().setS3ConfigOptions(config);
+ // S3A on Hadoop 3.4 keeps zero-byte directory markers by default
+ // (fs.s3a.directory.marker.retention=keep); pin the pre-3.4 "delete" policy so the raw
+ // listings below contain only actual result-store entries. The Presto subclass maps this
+ // key to presto.s3.directory.marker.retention, which PrestoS3FileSystem ignores.
+ config.setString("s3.directory.marker.retention", "delete");
+
// JobResultStore configuration
config.set(JobResultStoreOptions.DELETE_ON_COMMIT, Boolean.FALSE);
config.set(
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProviderTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProviderTest.java
index 43f3eb528f56fd..b5d231cc1b5dd3 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProviderTest.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenProviderTest.java
@@ -19,10 +19,14 @@
package org.apache.flink.fs.s3.common.token;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.security.token.DelegationTokenProvider.ObtainedDelegationTokens;
+import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.Optional;
+
import static org.apache.flink.core.security.token.DelegationTokenProvider.CONFIG_PREFIX;
import static org.assertj.core.api.Assertions.assertThat;
@@ -32,18 +36,37 @@ class AbstractS3DelegationTokenProviderTest {
private static final String REGION = "testRegion";
private static final String ACCESS_KEY_ID = "testAccessKeyId";
private static final String SECRET_ACCESS_KEY = "testSecretAccessKey";
+ private static final String SESSION_TOKEN = "testSessionToken";
+ private static final long EXPIRATION_EPOCH_MILLI = 1234567890L;
+
+ /** Records the arguments of the STS call and returns fixed session credentials. */
+ private static class TestS3DelegationTokenProvider extends AbstractS3DelegationTokenProvider {
+
+ private String seenRegion;
+ private String seenAccessKey;
+ private String seenSecretKey;
- private AbstractS3DelegationTokenProvider provider;
+ @Override
+ public String serviceName() {
+ return "s3";
+ }
+
+ @Override
+ protected S3SessionCredentials getSessionCredentials(
+ String region, String accessKey, String secretKey) {
+ this.seenRegion = region;
+ this.seenAccessKey = accessKey;
+ this.seenSecretKey = secretKey;
+ return new S3SessionCredentials(
+ ACCESS_KEY_ID, SECRET_ACCESS_KEY, SESSION_TOKEN, EXPIRATION_EPOCH_MILLI);
+ }
+ }
+
+ private TestS3DelegationTokenProvider provider;
@BeforeEach
void beforeEach() {
- provider =
- new AbstractS3DelegationTokenProvider() {
- @Override
- public String serviceName() {
- return "s3";
- }
- };
+ provider = new TestS3DelegationTokenProvider();
}
@Test
@@ -54,12 +77,36 @@ void delegationTokensRequiredShouldReturnFalseWithoutCredentials() {
@Test
void delegationTokensRequiredShouldReturnTrueWithCredentials() {
+ provider.init(createConfiguration());
+
+ assertThat(provider.delegationTokensRequired()).isTrue();
+ }
+
+ @Test
+ void obtainDelegationTokensShouldSerializeSessionCredentials() throws Exception {
+ provider.init(createConfiguration());
+
+ ObtainedDelegationTokens tokens = provider.obtainDelegationTokens();
+
+ assertThat(provider.seenRegion).isEqualTo(REGION);
+ assertThat(provider.seenAccessKey).isEqualTo(ACCESS_KEY_ID);
+ assertThat(provider.seenSecretKey).isEqualTo(SECRET_ACCESS_KEY);
+ assertThat(tokens.getValidUntil()).isEqualTo(Optional.of(EXPIRATION_EPOCH_MILLI));
+
+ S3SessionCredentials credentials =
+ InstantiationUtil.deserializeObject(
+ tokens.getTokens(), getClass().getClassLoader());
+ assertThat(credentials.getAccessKeyId()).isEqualTo(ACCESS_KEY_ID);
+ assertThat(credentials.getSecretAccessKey()).isEqualTo(SECRET_ACCESS_KEY);
+ assertThat(credentials.getSessionToken()).isEqualTo(SESSION_TOKEN);
+ assertThat(credentials.getExpirationEpochMilli()).isEqualTo(EXPIRATION_EPOCH_MILLI);
+ }
+
+ private static Configuration createConfiguration() {
Configuration configuration = new Configuration();
configuration.setString(CONFIG_PREFIX + ".s3.region", REGION);
configuration.setString(CONFIG_PREFIX + ".s3.access-key", ACCESS_KEY_ID);
configuration.setString(CONFIG_PREFIX + ".s3.secret-key", SECRET_ACCESS_KEY);
- provider.init(configuration);
-
- assertThat(provider.delegationTokensRequired()).isTrue();
+ return configuration;
}
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiverTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiverTest.java
index 5e579fda8cdeb1..e0b392ca5676b8 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiverTest.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/AbstractS3DelegationTokenReceiverTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.fs.s3.common.token;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.util.InstantiationUtil;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -32,16 +33,20 @@
class AbstractS3DelegationTokenReceiverTest {
private static final String PROVIDER_CLASS_NAME = "TestProvider";
+ private static final String SDK_V2_PROVIDER_CLASS_NAME =
+ "org.apache.flink.fs.s3.common.token.HadoopDynamicTemporaryAWSCredentialsProvider";
private static final String REGION = "testRegion";
@BeforeEach
void beforeEach() {
AbstractS3DelegationTokenReceiver.region = null;
+ AbstractS3DelegationTokenReceiver.credentials = null;
}
@AfterEach
void afterEach() {
AbstractS3DelegationTokenReceiver.region = null;
+ AbstractS3DelegationTokenReceiver.credentials = null;
}
@Test
@@ -76,6 +81,46 @@ void updateHadoopConfigShouldNotAddProviderWhenAlreadyExists() {
.isEqualTo(DynamicTemporaryAWSCredentialsProvider.NAME);
}
+ @Test
+ void updateHadoopConfigShouldSetGivenProviderName() {
+ org.apache.hadoop.conf.Configuration hadoopConfiguration =
+ new org.apache.hadoop.conf.Configuration();
+ hadoopConfiguration.set(PROVIDER_CONFIG_NAME, "");
+ AbstractS3DelegationTokenReceiver.updateHadoopConfig(
+ hadoopConfiguration, SDK_V2_PROVIDER_CLASS_NAME);
+ assertThat(hadoopConfiguration.get(PROVIDER_CONFIG_NAME))
+ .isEqualTo(SDK_V2_PROVIDER_CLASS_NAME);
+ }
+
+ @Test
+ void updateHadoopConfigShouldRemapLegacyProviderName() {
+ // A user-configured reference to the SDK v1 provider must be remapped in plugins that
+ // register a different (SDK v2) provider, because the v1 provider class cannot be loaded
+ // there.
+ org.apache.hadoop.conf.Configuration hadoopConfiguration =
+ new org.apache.hadoop.conf.Configuration();
+ hadoopConfiguration.set(
+ PROVIDER_CONFIG_NAME,
+ DynamicTemporaryAWSCredentialsProvider.NAME + "," + PROVIDER_CLASS_NAME);
+ AbstractS3DelegationTokenReceiver.updateHadoopConfig(
+ hadoopConfiguration, SDK_V2_PROVIDER_CLASS_NAME);
+ assertThat(hadoopConfiguration.get(PROVIDER_CONFIG_NAME))
+ .isEqualTo(SDK_V2_PROVIDER_CLASS_NAME + "," + PROVIDER_CLASS_NAME);
+ }
+
+ @Test
+ void updateHadoopConfigShouldDropDuplicateAfterRemapping() {
+ org.apache.hadoop.conf.Configuration hadoopConfiguration =
+ new org.apache.hadoop.conf.Configuration();
+ hadoopConfiguration.set(
+ PROVIDER_CONFIG_NAME,
+ SDK_V2_PROVIDER_CLASS_NAME + "," + DynamicTemporaryAWSCredentialsProvider.NAME);
+ AbstractS3DelegationTokenReceiver.updateHadoopConfig(
+ hadoopConfiguration, SDK_V2_PROVIDER_CLASS_NAME);
+ assertThat(hadoopConfiguration.get(PROVIDER_CONFIG_NAME))
+ .isEqualTo(SDK_V2_PROVIDER_CLASS_NAME);
+ }
+
@Test
void updateHadoopConfigShouldNotUpdateRegionWhenNotConfigured() {
AbstractS3DelegationTokenReceiver receiver = createReceiver();
@@ -100,6 +145,22 @@ void updateHadoopConfigShouldUpdateRegionWhenConfigured() {
assertThat(hadoopConfiguration.get("fs.s3a.endpoint.region")).isEqualTo(REGION);
}
+ @Test
+ void onNewTokensObtainedShouldStoreDeserializedCredentials() throws Exception {
+ AbstractS3DelegationTokenReceiver receiver = createReceiver();
+ S3SessionCredentials credentials =
+ new S3SessionCredentials("accessKeyId", "secretAccessKey", "sessionToken", 42L);
+
+ receiver.onNewTokensObtained(InstantiationUtil.serializeObject(credentials));
+
+ S3SessionCredentials storedCredentials = AbstractS3DelegationTokenReceiver.getCredentials();
+ assertThat(storedCredentials).isNotNull();
+ assertThat(storedCredentials.getAccessKeyId()).isEqualTo("accessKeyId");
+ assertThat(storedCredentials.getSecretAccessKey()).isEqualTo("secretAccessKey");
+ assertThat(storedCredentials.getSessionToken()).isEqualTo("sessionToken");
+ assertThat(storedCredentials.getExpirationEpochMilli()).isEqualTo(42L);
+ }
+
private AbstractS3DelegationTokenReceiver createReceiver() {
return new AbstractS3DelegationTokenReceiver() {
@Override
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProviderTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProviderTest.java
index 2a6c18a1024bf3..466b521f6e445b 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProviderTest.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/DynamicTemporaryAWSCredentialsProviderTest.java
@@ -20,9 +20,9 @@
import org.apache.flink.util.InstantiationUtil;
+import com.amazonaws.SdkClientException;
+import com.amazonaws.auth.AWSCredentials;
import com.amazonaws.auth.BasicSessionCredentials;
-import com.amazonaws.services.securitytoken.model.Credentials;
-import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -36,6 +36,7 @@ class DynamicTemporaryAWSCredentialsProviderTest {
private static final String ACCESS_KEY_ID = "testAccessKeyId";
private static final String SECRET_ACCESS_KEY = "testSecretAccessKey";
private static final String SESSION_TOKEN = "testSessionToken";
+ private static final long EXPIRATION_EPOCH_MILLI = 1234567890L;
@BeforeEach
void beforeEach() {
@@ -48,19 +49,32 @@ void afterEach() {
}
@Test
- void getCredentialsShouldThrowExceptionWhenNoCredentials() {
+ void nameMustMatchClassName() {
+ // NAME is a string literal so that referencing it never class-loads this provider (it
+ // implements an SDK v1 interface absent from the flink-s3-fs-hadoop jar); this pins the
+ // literal to the actual class name, which users reference from
+ // fs.s3a.aws.credentials.provider.
+ assertThat(DynamicTemporaryAWSCredentialsProvider.NAME)
+ .isEqualTo(DynamicTemporaryAWSCredentialsProvider.class.getName());
+ }
+
+ @Test
+ void getCredentialsShouldThrowSdkV1ExceptionWhenNoCredentials() {
DynamicTemporaryAWSCredentialsProvider provider =
new DynamicTemporaryAWSCredentialsProvider();
- assertThatThrownBy(provider::getCredentials).isInstanceOf(NoAwsCredentialsException.class);
+ // Must be the SDK v1 exception: Hadoop's NoAwsCredentialsException is based on the AWS SDK
+ // v2 exception hierarchy since Hadoop 3.4 and cannot be loaded in the presto plugin.
+ assertThatThrownBy(provider::getCredentials).isInstanceOf(SdkClientException.class);
}
@Test
- void getCredentialsShouldStoreCredentialsWhenCredentialsProvided() throws Exception {
+ void getCredentialsShouldReturnSessionCredentialsWhenProvided() throws Exception {
DynamicTemporaryAWSCredentialsProvider provider =
new DynamicTemporaryAWSCredentialsProvider();
- Credentials credentials =
- new Credentials(ACCESS_KEY_ID, SECRET_ACCESS_KEY, SESSION_TOKEN, null);
+ S3SessionCredentials credentials =
+ new S3SessionCredentials(
+ ACCESS_KEY_ID, SECRET_ACCESS_KEY, SESSION_TOKEN, EXPIRATION_EPOCH_MILLI);
AbstractS3DelegationTokenReceiver receiver =
new AbstractS3DelegationTokenReceiver() {
@Override
@@ -70,11 +84,12 @@ public String serviceName() {
};
receiver.onNewTokensObtained(InstantiationUtil.serializeObject(credentials));
- BasicSessionCredentials returnedCredentials =
- (BasicSessionCredentials) provider.getCredentials();
- assertThat(returnedCredentials.getAWSAccessKeyId()).isEqualTo(credentials.getAccessKeyId());
- assertThat(returnedCredentials.getAWSSecretKey())
- .isEqualTo(credentials.getSecretAccessKey());
- assertThat(returnedCredentials.getSessionToken()).isEqualTo(credentials.getSessionToken());
+
+ AWSCredentials v1Credentials = provider.getCredentials();
+ assertThat(v1Credentials).isInstanceOf(BasicSessionCredentials.class);
+ BasicSessionCredentials sessionCredentials = (BasicSessionCredentials) v1Credentials;
+ assertThat(sessionCredentials.getAWSAccessKeyId()).isEqualTo(ACCESS_KEY_ID);
+ assertThat(sessionCredentials.getAWSSecretKey()).isEqualTo(SECRET_ACCESS_KEY);
+ assertThat(sessionCredentials.getSessionToken()).isEqualTo(SESSION_TOKEN);
}
}
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/TokenSdkIsolationTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/TokenSdkIsolationTest.java
new file mode 100644
index 00000000000000..68b0e8309c3091
--- /dev/null
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/token/TokenSdkIsolationTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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.flink.fs.s3.common.token;
+
+import com.tngtech.archunit.core.domain.JavaClasses;
+import com.tngtech.archunit.core.importer.ClassFileImporter;
+import com.tngtech.archunit.core.importer.ImportOption;
+import org.junit.jupiter.api.Test;
+
+import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.classes;
+
+/**
+ * Guards the AWS SDK isolation of the shared classes in {@code flink-s3-fs-base}.
+ *
+ * The classes bundled into both S3 filesystem plugins must not reference AWS SDK types: the
+ * {@code flink-s3-fs-presto} jar ships only AWS SDK v1 and the {@code flink-s3-fs-hadoop} jar only
+ * AWS SDK v2, so a stray dependency turns into {@code NoClassDefFoundError} at runtime in one of
+ * the plugins — invisible to unit tests, whose classpath contains both SDKs.
+ */
+class TokenSdkIsolationTest {
+
+ private static final JavaClasses PRODUCTION_CLASSES =
+ new ClassFileImporter()
+ .withImportOption(new ImportOption.DoNotIncludeTests())
+ .importPackages("org.apache.flink.fs.s3.common");
+
+ @Test
+ void tokenClassesMustStaySdkAgnostic() {
+ classes()
+ .that()
+ .resideInAPackage("org.apache.flink.fs.s3.common.token")
+ .and()
+ .doNotHaveFullyQualifiedName(DynamicTemporaryAWSCredentialsProvider.NAME)
+ .should()
+ .onlyDependOnClassesThat()
+ .resideOutsideOfPackages("software.amazon.awssdk..", "com.amazonaws..")
+ .because(
+ "token state is shared between the SDK v1 presto plugin and the SDK v2 "
+ + "hadoop plugin")
+ .check(PRODUCTION_CLASSES);
+ }
+
+ @Test
+ void sdkV1CredentialsProviderMustNotDependOnSdkV2OrHadoopS3a() {
+ classes()
+ .that()
+ .haveFullyQualifiedName(DynamicTemporaryAWSCredentialsProvider.NAME)
+ .should()
+ .onlyDependOnClassesThat()
+ .resideOutsideOfPackages("software.amazon.awssdk..", "org.apache.hadoop.fs.s3a..")
+ .because(
+ "the provider is loaded inside the presto plugin where SDK v2 is absent, "
+ + "and Hadoop 3.4's s3a exception hierarchy is based on SDK v2")
+ .check(PRODUCTION_CLASSES);
+ }
+
+ @Test
+ void flinkS3FileSystemMustStaySdkAgnostic() {
+ classes()
+ .that()
+ .haveNameMatching(
+ "org\\.apache\\.flink\\.fs\\.s3\\.common\\.FlinkS3FileSystem(\\$.*)?")
+ .should()
+ .onlyDependOnClassesThat()
+ .resideOutsideOfPackages("software.amazon.awssdk..", "com.amazonaws..")
+ .because(
+ "the s5cmd credential lookup runs in both plugins and must not touch "
+ + "SDK-specific credential types")
+ .check(PRODUCTION_CLASSES);
+ }
+}
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImplTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImplTest.java
index 7ec0101d2e95e5..654f5a1e808fd5 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImplTest.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/RecoverableMultiPartUploadImplTest.java
@@ -23,14 +23,14 @@
import org.apache.flink.util.IOUtils;
import org.apache.flink.util.MathUtils;
-import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
-import com.amazonaws.services.s3.model.ObjectMetadata;
-import com.amazonaws.services.s3.model.PartETag;
-import com.amazonaws.services.s3.model.PutObjectResult;
-import com.amazonaws.services.s3.model.UploadPartResult;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import software.amazon.awssdk.services.s3.model.CompleteMultipartUploadResponse;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
+import software.amazon.awssdk.services.s3.model.PutObjectResponse;
+import software.amazon.awssdk.services.s3.model.UploadPartResponse;
import java.io.File;
import java.io.FileInputStream;
@@ -141,6 +141,67 @@ void uploadingNonClosedFileAsCompleteShouldThroughException() throws IOException
.isInstanceOf(IllegalStateException.class);
}
+ @Test
+ void partChecksumsShouldBePropagatedToRecoverable() throws IOException {
+ stubMultiPartUploader.enableChecksumGeneration();
+ final byte[] part = bytesOf("hello world");
+
+ uploadPart(part);
+ final S3Recoverable recoverable = multiPartUploadUnderTest.snapshotAndGetRecoverable(null);
+
+ assertThat(recoverable.parts()).hasSize(1);
+ final CompletedPart completedPart = recoverable.parts().get(0);
+ assertThat(completedPart.partNumber()).isEqualTo(1);
+ assertThat(completedPart.checksumCRC32C()).isEqualTo(createCRC32CChecksum(1));
+ }
+
+ @Test
+ void completedPartFromResponseShouldCopyEveryChecksum() {
+ final UploadPartResponse response =
+ UploadPartResponse.builder()
+ .eTag("etag")
+ .checksumCRC32("crc32")
+ .checksumCRC32C("crc32c")
+ .checksumCRC64NVME("crc64nvme")
+ .checksumSHA1("sha1")
+ .checksumSHA256("sha256")
+ .checksumSHA512("sha512")
+ .checksumMD5("md5")
+ .checksumXXHASH64("xxhash64")
+ .checksumXXHASH3("xxhash3")
+ .checksumXXHASH128("xxhash128")
+ .build();
+
+ final CompletedPart part =
+ RecoverableMultiPartUploadImpl.completedPartFromResponse(7, response);
+
+ assertThat(part.partNumber()).isEqualTo(7);
+ assertThat(part.eTag()).isEqualTo("etag");
+ assertThat(part.checksumCRC32()).isEqualTo("crc32");
+ assertThat(part.checksumCRC32C()).isEqualTo("crc32c");
+ assertThat(part.checksumCRC64NVME()).isEqualTo("crc64nvme");
+ assertThat(part.checksumSHA1()).isEqualTo("sha1");
+ assertThat(part.checksumSHA256()).isEqualTo("sha256");
+ assertThat(part.checksumSHA512()).isEqualTo("sha512");
+ assertThat(part.checksumMD5()).isEqualTo("md5");
+ assertThat(part.checksumXXHASH64()).isEqualTo("xxhash64");
+ assertThat(part.checksumXXHASH3()).isEqualTo("xxhash3");
+ assertThat(part.checksumXXHASH128()).isEqualTo("xxhash128");
+ }
+
+ @Test
+ void completedPartFromResponseShouldLeaveAbsentChecksumsUnset() {
+ final CompletedPart part =
+ RecoverableMultiPartUploadImpl.completedPartFromResponse(
+ 3, UploadPartResponse.builder().eTag("etag").build());
+
+ assertThat(part.partNumber()).isEqualTo(3);
+ assertThat(part.eTag()).isEqualTo("etag");
+ for (S3PartChecksum checksum : S3PartChecksum.values()) {
+ assertThat(checksum.valueOf(part)).isNull();
+ }
+ }
+
private static void assertThatHasMultiPartUploadWithPart(
StubMultiPartUploader actual, byte[] content, int partNo) {
TestUploadPartResult expectedCompletePart =
@@ -168,8 +229,8 @@ private static void assertThatIsEqualTo(
assertThat(actualRecoverable.incompleteObjectLength())
.isEqualTo(expectedRecoverable.incompleteObjectLength());
- assertThat(actualRecoverable.parts().stream().map(PartETag::getETag).toArray())
- .isEqualTo(expectedRecoverable.parts().stream().map(PartETag::getETag).toArray());
+ assertThat(actualRecoverable.parts().stream().map(CompletedPart::eTag).toArray())
+ .isEqualTo(expectedRecoverable.parts().stream().map(CompletedPart::eTag).toArray());
}
// ---------------------------------- Test Methods -------------------------------------------
@@ -180,12 +241,16 @@ private static byte[] bytesOf(String str) {
private static S3Recoverable createS3Recoverable(
byte[] incompletePart, byte[]... completeParts) {
- final List eTags = new ArrayList<>();
+ final List parts = new ArrayList<>();
int index = 1;
long bytesInPart = 0L;
for (byte[] part : completeParts) {
- eTags.add(new PartETag(index, createETag(TEST_OBJECT_NAME, index)));
+ parts.add(
+ CompletedPart.builder()
+ .partNumber(index)
+ .eTag(createETag(TEST_OBJECT_NAME, index))
+ .build());
bytesInPart += part.length;
index++;
}
@@ -193,7 +258,7 @@ private static S3Recoverable createS3Recoverable(
return new S3Recoverable(
TEST_OBJECT_NAME,
createMPUploadId(TEST_OBJECT_NAME),
- eTags,
+ parts,
bytesInPart,
"IGNORED-DUE-TO-RANDOMNESS",
(long) incompletePart.length);
@@ -201,21 +266,14 @@ private static S3Recoverable createS3Recoverable(
private static RecoverableMultiPartUploadImplTest.TestPutObjectResult createPutObjectResult(
String key, byte[] content) {
- final RecoverableMultiPartUploadImplTest.TestPutObjectResult result =
- new RecoverableMultiPartUploadImplTest.TestPutObjectResult();
- result.setETag(createETag(key, -1));
- result.setContent(content);
- return result;
+ return new RecoverableMultiPartUploadImplTest.TestPutObjectResult(
+ createETag(key, -1), content);
}
private static RecoverableMultiPartUploadImplTest.TestUploadPartResult createUploadPartResult(
String key, int number, byte[] payload) {
- final RecoverableMultiPartUploadImplTest.TestUploadPartResult result =
- new RecoverableMultiPartUploadImplTest.TestUploadPartResult();
- result.setETag(createETag(key, number));
- result.setPartNumber(number);
- result.setContent(payload);
- return result;
+ return new RecoverableMultiPartUploadImplTest.TestUploadPartResult(
+ number, createETag(key, number), payload);
}
private static String createMPUploadId(String key) {
@@ -226,6 +284,10 @@ private static String createETag(String key, int partNo) {
return "ETAG-" + key + '-' + partNo;
}
+ private static String createCRC32CChecksum(int partNo) {
+ return "CRC32C-" + partNo;
+ }
+
private S3Recoverable uploadObject(byte[] content) throws IOException {
final RefCountedBufferingFileStream incompletePartFile = writeContent(content);
incompletePartFile.flush();
@@ -283,6 +345,16 @@ private static class StubMultiPartUploader implements S3AccessHelper {
private final List
incompletePartsUploaded = new ArrayList<>();
+ private boolean checksumGenerationEnabled;
+
+ /**
+ * Makes upload-part responses carry a per-part checksum, as S3 does when {@code
+ * fs.s3a.checksum.generation} is enabled.
+ */
+ void enableChecksumGeneration() {
+ this.checksumGenerationEnabled = true;
+ }
+
List getCompletePartsUploaded() {
return completePartsUploaded;
}
@@ -297,7 +369,7 @@ public String startMultiPartUpload(String key) throws IOException {
}
@Override
- public UploadPartResult uploadPart(
+ public UploadPartResponse uploadPart(
String key, String uploadId, int partNumber, File inputFile, long length)
throws IOException {
final byte[] content =
@@ -306,7 +378,7 @@ public UploadPartResult uploadPart(
}
@Override
- public PutObjectResult putObject(String key, File inputFile) throws IOException {
+ public PutObjectResponse putObject(String key, File inputFile) throws IOException {
final byte[] content =
getFileContentBytes(inputFile, MathUtils.checkedDownCast(inputFile.length()));
return storeAndGetPutObjectResult(key, content);
@@ -323,10 +395,10 @@ public long getObject(String key, File targetLocation) throws IOException {
}
@Override
- public CompleteMultipartUploadResult commitMultiPartUpload(
+ public CompleteMultipartUploadResponse commitMultiPartUpload(
String key,
String uploadId,
- List partETags,
+ List partETags,
long length,
AtomicInteger errorCount)
throws IOException {
@@ -334,7 +406,7 @@ public CompleteMultipartUploadResult commitMultiPartUpload(
}
@Override
- public ObjectMetadata getObjectMetadata(String key) throws IOException {
+ public HeadObjectResponse getObjectMetadata(String key) throws IOException {
throw new UnsupportedOperationException();
}
@@ -344,37 +416,52 @@ private byte[] getFileContentBytes(File file, int length) throws IOException {
return content;
}
- private RecoverableMultiPartUploadImplTest.TestUploadPartResult storeAndGetUploadPartResult(
+ private UploadPartResponse storeAndGetUploadPartResult(
String key, int number, byte[] payload) {
final RecoverableMultiPartUploadImplTest.TestUploadPartResult result =
createUploadPartResult(key, number, payload);
completePartsUploaded.add(result);
- return result;
+ if (checksumGenerationEnabled) {
+ return result.toUploadPartResponse().toBuilder()
+ .checksumCRC32C(createCRC32CChecksum(number))
+ .build();
+ }
+ return result.toUploadPartResponse();
}
- private RecoverableMultiPartUploadImplTest.TestPutObjectResult storeAndGetPutObjectResult(
- String key, byte[] payload) {
+ private PutObjectResponse storeAndGetPutObjectResult(String key, byte[] payload) {
final RecoverableMultiPartUploadImplTest.TestPutObjectResult result =
createPutObjectResult(key, payload);
incompletePartsUploaded.add(result);
- return result;
+ return result.toPutObjectResponse();
}
}
- /** A {@link PutObjectResult} that also contains the actual content of the uploaded part. */
- private static class TestPutObjectResult extends PutObjectResult {
- private static final long serialVersionUID = 1L;
+ /**
+ * A wrapper for {@link PutObjectResponse} that also contains the actual content of the uploaded
+ * part.
+ */
+ private static class TestPutObjectResult {
+ private final String eTag;
+ private final byte[] content;
- private byte[] content;
+ TestPutObjectResult(String eTag, byte[] content) {
+ this.eTag = eTag;
+ this.content = content;
+ }
- void setContent(byte[] payload) {
- this.content = payload;
+ public String getETag() {
+ return eTag;
}
public byte[] getContent() {
return content;
}
+ public PutObjectResponse toPutObjectResponse() {
+ return PutObjectResponse.builder().eTag(eTag).build();
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -400,21 +487,37 @@ public String toString() {
}
}
- /** A {@link UploadPartResult} that also contains the actual content of the uploaded part. */
- private static class TestUploadPartResult extends UploadPartResult {
-
- private static final long serialVersionUID = 1L;
+ /**
+ * A wrapper for {@link UploadPartResponse} that also contains the actual content of the
+ * uploaded part.
+ */
+ private static class TestUploadPartResult {
+ private final int partNumber;
+ private final String eTag;
+ private final byte[] content;
+
+ TestUploadPartResult(int partNumber, String eTag, byte[] content) {
+ this.partNumber = partNumber;
+ this.eTag = eTag;
+ this.content = content;
+ }
- private byte[] content;
+ public String getETag() {
+ return eTag;
+ }
- void setContent(byte[] content) {
- this.content = content;
+ public int getPartNumber() {
+ return partNumber;
}
public byte[] getContent() {
return content;
}
+ public UploadPartResponse toUploadPartResponse() {
+ return UploadPartResponse.builder().eTag(eTag).build();
+ }
+
@Override
public boolean equals(Object o) {
if (this == o) {
diff --git a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializerTest.java b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializerTest.java
index 58f7697f1bdb19..dc970a74b138c0 100644
--- a/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializerTest.java
+++ b/flink-filesystems/flink-s3-fs-base/src/test/java/org/apache/flink/fs/s3/common/writer/S3RecoverableSerializerTest.java
@@ -18,14 +18,20 @@
package org.apache.flink.fs.s3.common.writer;
-import com.amazonaws.services.s3.model.PartETag;
import org.junit.jupiter.api.Test;
+import software.amazon.awssdk.services.s3.model.CompletedPart;
+
+import javax.annotation.Nullable;
import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
/** Tests for the {@link S3RecoverableSerializer}. */
class S3RecoverableSerializerTest {
@@ -40,12 +46,27 @@ class S3RecoverableSerializerTest {
private static final String ETAG_PREFIX = "TEST-ETAG-";
+ private static final String CRC32C_PREFIX = "TEST-CRC32C-";
+
+ private static final String SHA256_PREFIX = "TEST-SHA256-";
+
+ /** Version-2 wire tags of the checksums used by the fixtures; pinned independently. */
+ private static final byte CRC32C_WIRE_TAG = 2;
+
+ private static final byte SHA256_WIRE_TAG = 5;
+
+ @Test
+ void serializerVersionIsTwo() {
+ assertThat(serializer.getVersion()).isEqualTo(2);
+ }
+
@Test
void serializeEmptyS3Recoverable() throws IOException {
S3Recoverable originalEmptyRecoverable = createTestS3Recoverable(false);
byte[] serializedRecoverable = serializer.serialize(originalEmptyRecoverable);
- S3Recoverable copiedEmptyRecoverable = serializer.deserialize(1, serializedRecoverable);
+ S3Recoverable copiedEmptyRecoverable =
+ serializer.deserialize(serializer.getVersion(), serializedRecoverable);
assertThatIsEqualTo(originalEmptyRecoverable, copiedEmptyRecoverable);
}
@@ -56,7 +77,7 @@ void serializeS3RecoverableWithoutIncompleteObject() throws IOException {
byte[] serializedRecoverable = serializer.serialize(originalNoIncompletePartRecoverable);
S3Recoverable copiedNoIncompletePartRecoverable =
- serializer.deserialize(1, serializedRecoverable);
+ serializer.deserialize(serializer.getVersion(), serializedRecoverable);
assertThatIsEqualTo(originalNoIncompletePartRecoverable, copiedNoIncompletePartRecoverable);
}
@@ -67,7 +88,7 @@ void serializeS3RecoverableOnlyWithIncompleteObject() throws IOException {
byte[] serializedRecoverable = serializer.serialize(originalOnlyIncompletePartRecoverable);
S3Recoverable copiedOnlyIncompletePartRecoverable =
- serializer.deserialize(1, serializedRecoverable);
+ serializer.deserialize(serializer.getVersion(), serializedRecoverable);
assertThatIsEqualTo(
originalOnlyIncompletePartRecoverable, copiedOnlyIncompletePartRecoverable);
@@ -78,11 +99,109 @@ void serializeS3RecoverableWithCompleteAndIncompleteParts() throws IOException {
S3Recoverable originalFullRecoverable = createTestS3Recoverable(true, 1, 5, 9);
byte[] serializedRecoverable = serializer.serialize(originalFullRecoverable);
- S3Recoverable copiedFullRecoverable = serializer.deserialize(1, serializedRecoverable);
+ S3Recoverable copiedFullRecoverable =
+ serializer.deserialize(serializer.getVersion(), serializedRecoverable);
assertThatIsEqualTo(originalFullRecoverable, copiedFullRecoverable);
}
+ @Test
+ void serializeS3RecoverableWithPartChecksums() throws IOException {
+ S3Recoverable originalChecksummedRecoverable =
+ createChecksummedTestS3Recoverable(true, 1, 5, 9);
+
+ byte[] serializedRecoverable = serializer.serialize(originalChecksummedRecoverable);
+ S3Recoverable copiedChecksummedRecoverable =
+ serializer.deserialize(serializer.getVersion(), serializedRecoverable);
+
+ assertThatIsEqualTo(originalChecksummedRecoverable, copiedChecksummedRecoverable);
+ }
+
+ // ------------------------------------------------------------------------
+ // Wire-format fixture tests. The serialized bytes are persisted in
+ // checkpoints and savepoints, so state written by previous releases must
+ // keep deserializing. These tests pin the exact byte layouts independently
+ // of the serializer implementation; if they fail, the layout changed and a
+ // new serializer version is required instead.
+ // ------------------------------------------------------------------------
+
+ @Test
+ void wireFormatV1StillDeserializes() throws IOException {
+ S3Recoverable expected = createTestS3Recoverable(true, 1, 5, 9);
+ byte[] v1Bytes =
+ buildV1WireBytes(
+ TEST_OBJECT_NAME,
+ TEST_UPLOAD_ID,
+ new int[] {1, 5, 9},
+ 12345L,
+ INCOMPLETE_OBJECT_NAME,
+ 54321L);
+
+ assertThatIsEqualTo(serializer.deserialize(1, v1Bytes), expected);
+ }
+
+ @Test
+ void wireFormatV1StillDeserializesWithoutIncompleteObject() throws IOException {
+ S3Recoverable expected = createTestS3Recoverable(false, 1, 5, 9);
+ byte[] v1Bytes =
+ buildV1WireBytes(
+ TEST_OBJECT_NAME, TEST_UPLOAD_ID, new int[] {1, 5, 9}, 12345L, null, -1L);
+
+ assertThatIsEqualTo(serializer.deserialize(1, v1Bytes), expected);
+ }
+
+ @Test
+ void wireFormatV2IsStableWithoutChecksums() throws IOException {
+ S3Recoverable recoverable = createTestS3Recoverable(true, 1, 5, 9);
+ byte[] expectedBytes =
+ buildV2WireBytes(
+ TEST_OBJECT_NAME,
+ TEST_UPLOAD_ID,
+ new int[] {1, 5, 9},
+ false,
+ 12345L,
+ INCOMPLETE_OBJECT_NAME,
+ 54321L);
+
+ assertThat(serializer.serialize(recoverable)).isEqualTo(expectedBytes);
+ assertThatIsEqualTo(serializer.deserialize(2, expectedBytes), recoverable);
+ }
+
+ @Test
+ void wireFormatV2IsStableWithChecksums() throws IOException {
+ S3Recoverable recoverable = createChecksummedTestS3Recoverable(false, 1, 5, 9);
+ byte[] expectedBytes =
+ buildV2WireBytes(
+ TEST_OBJECT_NAME,
+ TEST_UPLOAD_ID,
+ new int[] {1, 5, 9},
+ true,
+ 12345L,
+ null,
+ -1L);
+
+ assertThat(serializer.serialize(recoverable)).isEqualTo(expectedBytes);
+ assertThatIsEqualTo(serializer.deserialize(2, expectedBytes), recoverable);
+ }
+
+ @Test
+ void unknownVersionIsRejected() throws IOException {
+ byte[] serialized = serializer.serialize(createTestS3Recoverable(false, 1));
+
+ assertThatThrownBy(() -> serializer.deserialize(3, serialized))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unrecognized version");
+ }
+
+ @Test
+ void unknownChecksumTagIsRejected() {
+ byte[] corruptBytes = buildV2WireBytesWithSingleChecksum((byte) 99, "SOME-VALUE");
+
+ assertThatThrownBy(() -> serializer.deserialize(2, corruptBytes))
+ .isInstanceOf(IOException.class)
+ .hasMessageContaining("Unknown part checksum tag");
+ }
+
private static void assertThatIsEqualTo(
S3Recoverable actualRecoverable, S3Recoverable expectedRecoverable) {
assertThat(actualRecoverable.getObjectName())
@@ -94,33 +213,220 @@ private static void assertThatIsEqualTo(
.isEqualTo(expectedRecoverable.incompleteObjectName());
assertThat(actualRecoverable.incompleteObjectLength())
.isEqualTo(expectedRecoverable.incompleteObjectLength());
- assertThat(actualRecoverable.parts().stream().map(PartETag::getETag).toArray())
- .isEqualTo(expectedRecoverable.parts().stream().map(PartETag::getETag).toArray());
+ // full CompletedPart equality, covering the checksum fields
+ assertThat(actualRecoverable.parts().toArray())
+ .isEqualTo(expectedRecoverable.parts().toArray());
}
// --------------------------------- Test Utils ---------------------------------
private static S3Recoverable createTestS3Recoverable(
boolean withIncompletePart, int... partNumbers) {
- List etags = new ArrayList<>();
+ return createTestS3Recoverable(false, withIncompletePart, partNumbers);
+ }
+
+ private static S3Recoverable createChecksummedTestS3Recoverable(
+ boolean withIncompletePart, int... partNumbers) {
+ return createTestS3Recoverable(true, withIncompletePart, partNumbers);
+ }
+
+ private static S3Recoverable createTestS3Recoverable(
+ boolean withChecksums, boolean withIncompletePart, int... partNumbers) {
+ List parts = new ArrayList<>();
for (int i : partNumbers) {
- etags.add(createEtag(i));
+ parts.add(createCompletedPart(i, withChecksums));
}
if (withIncompletePart) {
return new S3Recoverable(
TEST_OBJECT_NAME,
TEST_UPLOAD_ID,
- etags,
+ parts,
12345L,
INCOMPLETE_OBJECT_NAME,
54321L);
} else {
- return new S3Recoverable(TEST_OBJECT_NAME, TEST_UPLOAD_ID, etags, 12345L);
+ return new S3Recoverable(TEST_OBJECT_NAME, TEST_UPLOAD_ID, parts, 12345L);
+ }
+ }
+
+ private static CompletedPart createCompletedPart(int partNumber, boolean withChecksums) {
+ CompletedPart.Builder builder =
+ CompletedPart.builder().partNumber(partNumber).eTag(ETAG_PREFIX + partNumber);
+ if (withChecksums) {
+ builder.checksumCRC32C(CRC32C_PREFIX + partNumber)
+ .checksumSHA256(SHA256_PREFIX + partNumber);
+ }
+ return builder.build();
+ }
+
+ /**
+ * Hand-builds the version-1 wire layout: little-endian; magic number (int); key length (int) +
+ * UTF-8 bytes; upload id length (int) + bytes; part count (int); per part: part number (int) +
+ * etag length (int) + bytes; bytes in parts (long); incomplete object name length (int, 0 for
+ * none) + bytes; incomplete object length (long).
+ */
+ private static byte[] buildV1WireBytes(
+ String objectName,
+ String uploadId,
+ int[] partNumbers,
+ long numBytesInParts,
+ @Nullable String incompleteObjectName,
+ long incompleteObjectLength) {
+ final byte[] keyBytes = objectName.getBytes(StandardCharsets.UTF_8);
+ final byte[] uploadIdBytes = uploadId.getBytes(StandardCharsets.UTF_8);
+ final byte[][] etags = new byte[partNumbers.length][];
+ int partBytes = 0;
+ for (int i = 0; i < partNumbers.length; i++) {
+ etags[i] = (ETAG_PREFIX + partNumbers[i]).getBytes(StandardCharsets.UTF_8);
+ partBytes += 2 * Integer.BYTES + etags[i].length;
+ }
+ final byte[] incompleteBytes =
+ incompleteObjectName == null
+ ? new byte[0]
+ : incompleteObjectName.getBytes(StandardCharsets.UTF_8);
+
+ final ByteBuffer bb =
+ ByteBuffer.allocate(
+ 4 * Integer.BYTES
+ + keyBytes.length
+ + uploadIdBytes.length
+ + partBytes
+ + 2 * Long.BYTES
+ + Integer.BYTES
+ + incompleteBytes.length)
+ .order(ByteOrder.LITTLE_ENDIAN);
+
+ bb.putInt(0x98761432);
+ bb.putInt(keyBytes.length);
+ bb.put(keyBytes);
+ bb.putInt(uploadIdBytes.length);
+ bb.put(uploadIdBytes);
+ bb.putInt(partNumbers.length);
+ for (int i = 0; i < partNumbers.length; i++) {
+ bb.putInt(partNumbers[i]);
+ bb.putInt(etags[i].length);
+ bb.put(etags[i]);
+ }
+ bb.putLong(numBytesInParts);
+ bb.putInt(incompleteBytes.length);
+ bb.put(incompleteBytes);
+ bb.putLong(incompleteObjectLength);
+
+ return bb.array();
+ }
+
+ /**
+ * Hand-builds the version-2 wire layout: like version 1, but each part is followed by its
+ * checksum count (byte) and per checksum the wire tag (byte) + value length (int) + UTF-8
+ * bytes. Parts built with {@code withChecksums} carry a CRC32C (tag 2) and a SHA256 (tag 5)
+ * checksum, in wire-tag order.
+ */
+ private static byte[] buildV2WireBytes(
+ String objectName,
+ String uploadId,
+ int[] partNumbers,
+ boolean withChecksums,
+ long numBytesInParts,
+ @Nullable String incompleteObjectName,
+ long incompleteObjectLength) {
+ final byte[] keyBytes = objectName.getBytes(StandardCharsets.UTF_8);
+ final byte[] uploadIdBytes = uploadId.getBytes(StandardCharsets.UTF_8);
+ final byte[][] etags = new byte[partNumbers.length][];
+ final byte[][] crc32cs = new byte[partNumbers.length][];
+ final byte[][] sha256s = new byte[partNumbers.length][];
+ int partBytes = 0;
+ for (int i = 0; i < partNumbers.length; i++) {
+ etags[i] = (ETAG_PREFIX + partNumbers[i]).getBytes(StandardCharsets.UTF_8);
+ partBytes += 2 * Integer.BYTES + etags[i].length + Byte.BYTES;
+ if (withChecksums) {
+ crc32cs[i] = (CRC32C_PREFIX + partNumbers[i]).getBytes(StandardCharsets.UTF_8);
+ sha256s[i] = (SHA256_PREFIX + partNumbers[i]).getBytes(StandardCharsets.UTF_8);
+ partBytes +=
+ 2 * (Byte.BYTES + Integer.BYTES) + crc32cs[i].length + sha256s[i].length;
+ }
+ }
+ final byte[] incompleteBytes =
+ incompleteObjectName == null
+ ? new byte[0]
+ : incompleteObjectName.getBytes(StandardCharsets.UTF_8);
+
+ final ByteBuffer bb =
+ ByteBuffer.allocate(
+ 4 * Integer.BYTES
+ + keyBytes.length
+ + uploadIdBytes.length
+ + partBytes
+ + 2 * Long.BYTES
+ + Integer.BYTES
+ + incompleteBytes.length)
+ .order(ByteOrder.LITTLE_ENDIAN);
+
+ bb.putInt(0x98761432);
+ bb.putInt(keyBytes.length);
+ bb.put(keyBytes);
+ bb.putInt(uploadIdBytes.length);
+ bb.put(uploadIdBytes);
+ bb.putInt(partNumbers.length);
+ for (int i = 0; i < partNumbers.length; i++) {
+ bb.putInt(partNumbers[i]);
+ bb.putInt(etags[i].length);
+ bb.put(etags[i]);
+ if (withChecksums) {
+ bb.put((byte) 2);
+ bb.put(CRC32C_WIRE_TAG);
+ bb.putInt(crc32cs[i].length);
+ bb.put(crc32cs[i]);
+ bb.put(SHA256_WIRE_TAG);
+ bb.putInt(sha256s[i].length);
+ bb.put(sha256s[i]);
+ } else {
+ bb.put((byte) 0);
+ }
}
+ bb.putLong(numBytesInParts);
+ bb.putInt(incompleteBytes.length);
+ bb.put(incompleteBytes);
+ bb.putLong(incompleteObjectLength);
+
+ return bb.array();
}
- private static PartETag createEtag(int partNumber) {
- return new PartETag(partNumber, ETAG_PREFIX + partNumber);
+ /** Builds version-2 bytes for a single one-part recoverable carrying one checksum entry. */
+ private static byte[] buildV2WireBytesWithSingleChecksum(byte wireTag, String checksumValue) {
+ final byte[] keyBytes = TEST_OBJECT_NAME.getBytes(StandardCharsets.UTF_8);
+ final byte[] uploadIdBytes = TEST_UPLOAD_ID.getBytes(StandardCharsets.UTF_8);
+ final byte[] etagBytes = (ETAG_PREFIX + 1).getBytes(StandardCharsets.UTF_8);
+ final byte[] checksumBytes = checksumValue.getBytes(StandardCharsets.UTF_8);
+
+ final ByteBuffer bb =
+ ByteBuffer.allocate(
+ 8 * Integer.BYTES
+ + keyBytes.length
+ + uploadIdBytes.length
+ + etagBytes.length
+ + 2 * Byte.BYTES
+ + checksumBytes.length
+ + 2 * Long.BYTES)
+ .order(ByteOrder.LITTLE_ENDIAN);
+
+ bb.putInt(0x98761432);
+ bb.putInt(keyBytes.length);
+ bb.put(keyBytes);
+ bb.putInt(uploadIdBytes.length);
+ bb.put(uploadIdBytes);
+ bb.putInt(1);
+ bb.putInt(1);
+ bb.putInt(etagBytes.length);
+ bb.put(etagBytes);
+ bb.put((byte) 1);
+ bb.put(wireTag);
+ bb.putInt(checksumBytes.length);
+ bb.put(checksumBytes);
+ bb.putLong(12345L);
+ bb.putInt(0);
+ bb.putLong(-1L);
+
+ return bb.array();
}
}
diff --git a/flink-filesystems/flink-s3-fs-hadoop/pom.xml b/flink-filesystems/flink-s3-fs-hadoop/pom.xml
index 098296e49561a7..98a6eea8cd5565 100644
--- a/flink-filesystems/flink-s3-fs-hadoop/pom.xml
+++ b/flink-filesystems/flink-s3-fs-hadoop/pom.xml
@@ -84,12 +84,43 @@ under the License.
test