You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2018/11/16 20:15:15 UTC
[flink] 03/13: [FLINK-10869] [tests] Update all S3 tests to use new
test credentials
This is an automated email from the ASF dual-hosted git repository.
sewen pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fe358d608a6a7c925b9918d1ac3adeef24454cef
Author: Stephan Ewen <se...@apache.org>
AuthorDate: Thu Nov 15 16:36:23 2018 +0100
[FLINK-10869] [tests] Update all S3 tests to use new test credentials
---
.../s3hadoop/HadoopS3FileSystemBehaviorITCase.java | 17 +---
.../fs/s3hadoop/HadoopS3FileSystemITCase.java | 44 ++++-----
.../HadoopS3RecoverableWriterExceptionTest.java | 20 ++--
.../fs/s3hadoop/HadoopS3RecoverableWriterTest.java | 20 ++--
.../s3presto/PrestoS3FileSystemBehaviorITCase.java | 17 +---
.../fs/s3presto/PrestoS3FileSystemITCase.java | 41 ++++----
.../fs/s3presto/PrestoS3RecoverableWriterTest.java | 21 ++--
.../apache/flink/testutils/s3/S3Credentials.java | 106 ---------------------
.../flink/yarn/YarnFileStageTestS3ITCase.java | 25 ++---
9 files changed, 76 insertions(+), 235 deletions(-)
diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemBehaviorITCase.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemBehaviorITCase.java
index c8aaaee..df73cda 100644
--- a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemBehaviorITCase.java
+++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemBehaviorITCase.java
@@ -23,9 +23,9 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemBehaviorTestSuite;
import org.apache.flink.core.fs.FileSystemKind;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.BeforeClass;
import java.io.IOException;
@@ -36,24 +36,17 @@ import java.util.UUID;
*/
public class HadoopS3FileSystemBehaviorITCase extends FileSystemBehaviorTestSuite {
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
-
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
}
@@ -69,7 +62,7 @@ public class HadoopS3FileSystemBehaviorITCase extends FileSystemBehaviorTestSuit
@Override
public Path getBasePath() throws Exception {
- return new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ return new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
}
@Override
diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemITCase.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemITCase.java
index 6dbdac5..2195bd0 100644
--- a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemITCase.java
+++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3FileSystemITCase.java
@@ -25,10 +25,10 @@ import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.apache.flink.util.TestLogger;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -69,10 +69,6 @@ public class HadoopS3FileSystemITCase extends TestLogger {
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
-
/**
* Will be updated by {@link #checkCredentialsAndSetup()} if the test is not skipped.
*/
@@ -81,18 +77,16 @@ public class HadoopS3FileSystemITCase extends TestLogger {
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
// check for uniqueness of the test directory
- final Path directory = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ final Path directory = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
final FileSystem fs = directory.getFileSystem();
// directory must not yet exist
@@ -108,11 +102,11 @@ public class HadoopS3FileSystemITCase extends TestLogger {
final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
- final Path directory = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ final Path directory = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
final FileSystem fs = directory.getFileSystem();
// clean up
@@ -128,7 +122,7 @@ public class HadoopS3FileSystemITCase extends TestLogger {
}
private String getBasePath() {
- return scheme + "://" + BUCKET + '/' + TEST_DATA_DIR + "/" + scheme;
+ return S3TestCredentials.getTestBucketUriWithScheme(scheme) + TEST_DATA_DIR + '/' + scheme;
}
@Test
@@ -138,8 +132,8 @@ public class HadoopS3FileSystemITCase extends TestLogger {
// standard Hadoop-style credential keys
{
Configuration conf = new Configuration();
- conf.setString("fs.s3a.access.key", ACCESS_KEY);
- conf.setString("fs.s3a.secret.key", SECRET_KEY);
+ conf.setString("fs.s3a.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("fs.s3a.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem();
@@ -148,8 +142,8 @@ public class HadoopS3FileSystemITCase extends TestLogger {
// shortened Hadoop-style credential keys
{
Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem();
@@ -158,8 +152,8 @@ public class HadoopS3FileSystemITCase extends TestLogger {
// shortened Presto-style credential keys
{
Configuration conf = new Configuration();
- conf.setString("s3.access-key", ACCESS_KEY);
- conf.setString("s3.secret-key", SECRET_KEY);
+ conf.setString("s3.access-key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret-key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem();
@@ -170,8 +164,8 @@ public class HadoopS3FileSystemITCase extends TestLogger {
public void testSimpleFileWriteAndRead() throws Exception {
final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
final String testLine = "Hello Upload!";
@@ -208,8 +202,8 @@ public class HadoopS3FileSystemITCase extends TestLogger {
public void testDirectoryListing() throws Exception {
final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterExceptionTest.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterExceptionTest.java
index 634fa00..3fb67c1 100644
--- a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterExceptionTest.java
+++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterExceptionTest.java
@@ -25,12 +25,12 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.fs.s3.common.FlinkS3FileSystem;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;
import org.junit.After;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -55,10 +55,6 @@ public class HadoopS3RecoverableWriterExceptionTest extends TestLogger {
// ----------------------- S3 general configuration -----------------------
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final long PART_UPLOAD_MIN_SIZE_VALUE = 7L << 20;
private static final int MAX_CONCURRENT_UPLOADS_VALUE = 2;
@@ -66,9 +62,7 @@ public class HadoopS3RecoverableWriterExceptionTest extends TestLogger {
private static final Random RND = new Random();
- private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
-
- private static final Path basePath = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ private static Path basePath;
private static FlinkS3FileSystem fileSystem;
@@ -89,14 +83,14 @@ public class HadoopS3RecoverableWriterExceptionTest extends TestLogger {
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
+
+ basePath = new Path(S3TestCredentials.getTestBucketUri() + "tests-" + UUID.randomUUID());
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
conf.setLong(PART_UPLOAD_MIN_SIZE, PART_UPLOAD_MIN_SIZE_VALUE);
conf.setInteger(MAX_CONCURRENT_UPLOADS, MAX_CONCURRENT_UPLOADS_VALUE);
diff --git a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterTest.java b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterTest.java
index 4a1368a..a9027a0 100644
--- a/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterTest.java
+++ b/flink-filesystems/flink-s3-fs-hadoop/src/test/java/org/apache/flink/fs/s3hadoop/HadoopS3RecoverableWriterTest.java
@@ -27,6 +27,7 @@ import org.apache.flink.core.fs.RecoverableFsDataOutputStream;
import org.apache.flink.core.fs.RecoverableWriter;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.fs.s3.common.FlinkS3FileSystem;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.apache.flink.util.MathUtils;
import org.apache.flink.util.StringUtils;
import org.apache.flink.util.TestLogger;
@@ -34,7 +35,6 @@ import org.apache.flink.util.TestLogger;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Assert;
-import org.junit.Assume;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
@@ -63,10 +63,6 @@ public class HadoopS3RecoverableWriterTest extends TestLogger {
// ----------------------- S3 general configuration -----------------------
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final long PART_UPLOAD_MIN_SIZE_VALUE = 7L << 20;
private static final int MAX_CONCURRENT_UPLOADS_VALUE = 2;
@@ -74,9 +70,7 @@ public class HadoopS3RecoverableWriterTest extends TestLogger {
private static final Random RND = new Random();
- private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
-
- private static final Path basePath = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ private static Path basePath;
private static FlinkS3FileSystem fileSystem;
@@ -101,14 +95,14 @@ public class HadoopS3RecoverableWriterTest extends TestLogger {
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
+
+ basePath = new Path(S3TestCredentials.getTestBucketUri() + "tests-" + UUID.randomUUID());
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
conf.setLong(PART_UPLOAD_MIN_SIZE, PART_UPLOAD_MIN_SIZE_VALUE);
conf.setInteger(MAX_CONCURRENT_UPLOADS, MAX_CONCURRENT_UPLOADS_VALUE);
diff --git a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemBehaviorITCase.java b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemBehaviorITCase.java
index 812404c..1603927 100644
--- a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemBehaviorITCase.java
+++ b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemBehaviorITCase.java
@@ -23,9 +23,9 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystemBehaviorTestSuite;
import org.apache.flink.core.fs.FileSystemKind;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.BeforeClass;
import java.io.IOException;
@@ -36,24 +36,17 @@ import java.util.UUID;
*/
public class PrestoS3FileSystemBehaviorITCase extends FileSystemBehaviorTestSuite {
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
-
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
}
@@ -69,7 +62,7 @@ public class PrestoS3FileSystemBehaviorITCase extends FileSystemBehaviorTestSuit
@Override
public Path getBasePath() throws Exception {
- return new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
+ return new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
}
@Override
diff --git a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemITCase.java b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemITCase.java
index cc5c993..9a008a1 100644
--- a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemITCase.java
+++ b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3FileSystemITCase.java
@@ -25,9 +25,9 @@ import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.FileSystem.WriteMode;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.apache.flink.util.TestLogger;
-import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -68,23 +68,16 @@ public class PrestoS3FileSystemITCase extends TestLogger {
return Arrays.asList("s3", "s3p");
}
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
-
@BeforeClass
public static void checkIfCredentialsArePresent() {
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
}
@Test
public void testConfigKeysForwarding() throws Exception {
- final Path path = new Path(scheme + "://" + BUCKET + '/' + TEST_DATA_DIR);
+ final Path path = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR);
// access without credentials should fail
{
@@ -103,8 +96,8 @@ public class PrestoS3FileSystemITCase extends TestLogger {
{
Configuration conf = new Configuration();
conf.setString(S3_USE_INSTANCE_CREDENTIALS, "false");
- conf.setString("presto.s3.access-key", ACCESS_KEY);
- conf.setString("presto.s3.secret-key", SECRET_KEY);
+ conf.setString("presto.s3.access-key", S3TestCredentials.getS3AccessKey());
+ conf.setString("presto.s3.secret-key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem().exists(path);
@@ -114,8 +107,8 @@ public class PrestoS3FileSystemITCase extends TestLogger {
{
Configuration conf = new Configuration();
conf.setString(S3_USE_INSTANCE_CREDENTIALS, "false");
- conf.setString("s3.access-key", ACCESS_KEY);
- conf.setString("s3.secret-key", SECRET_KEY);
+ conf.setString("s3.access-key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret-key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem().exists(path);
@@ -125,8 +118,8 @@ public class PrestoS3FileSystemITCase extends TestLogger {
{
Configuration conf = new Configuration();
conf.setString(S3_USE_INSTANCE_CREDENTIALS, "false");
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem().exists(path);
@@ -136,8 +129,8 @@ public class PrestoS3FileSystemITCase extends TestLogger {
{
Configuration conf = new Configuration();
conf.setString(S3_USE_INSTANCE_CREDENTIALS, "false");
- conf.setString("presto.s3.access.key", ACCESS_KEY);
- conf.setString("presto.s3.secret.key", SECRET_KEY);
+ conf.setString("presto.s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("presto.s3.secret.key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
path.getFileSystem().exists(path);
@@ -151,14 +144,14 @@ public class PrestoS3FileSystemITCase extends TestLogger {
public void testSimpleFileWriteAndRead() throws Exception {
final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs
final Configuration conf = new Configuration();
- conf.setString("s3.access-key", ACCESS_KEY);
- conf.setString("s3.secret-key", SECRET_KEY);
+ conf.setString("s3.access-key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret-key", S3TestCredentials.getS3SecretKey());
final String testLine = "Hello Upload!";
FileSystem.initialize(conf);
- final Path path = new Path(scheme + "://" + BUCKET + '/' + TEST_DATA_DIR + "/test.txt");
+ final Path path = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR + "/test.txt");
final FileSystem fs = path.getFileSystem();
try {
@@ -186,12 +179,12 @@ public class PrestoS3FileSystemITCase extends TestLogger {
public void testDirectoryListing() throws Exception {
final long deadline = System.nanoTime() + 30_000_000_000L; // 30 secs
final Configuration conf = new Configuration();
- conf.setString("s3.access-key", ACCESS_KEY);
- conf.setString("s3.secret-key", SECRET_KEY);
+ conf.setString("s3.access-key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret-key", S3TestCredentials.getS3SecretKey());
FileSystem.initialize(conf);
- final Path directory = new Path(scheme + "://" + BUCKET + '/' + TEST_DATA_DIR + "/testdir/");
+ final Path directory = new Path(S3TestCredentials.getTestBucketUri() + TEST_DATA_DIR + "/testdir/");
final FileSystem fs = directory.getFileSystem();
// directory must not yet exist
diff --git a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3RecoverableWriterTest.java b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3RecoverableWriterTest.java
index 580d957..59d2f16 100644
--- a/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3RecoverableWriterTest.java
+++ b/flink-filesystems/flink-s3-fs-presto/src/test/java/org/apache/flink/fs/s3presto/PrestoS3RecoverableWriterTest.java
@@ -21,15 +21,15 @@ package org.apache.flink.fs.s3presto;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.CoreOptions;
import org.apache.flink.core.fs.FileSystem;
-import org.apache.flink.core.fs.Path;
import org.apache.flink.fs.s3.common.FlinkS3FileSystem;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.Test;
import java.io.IOException;
+import java.net.URI;
import java.util.UUID;
import static org.apache.flink.fs.s3.common.AbstractS3FileSystemFactory.MAX_CONCURRENT_UPLOADS;
@@ -42,10 +42,6 @@ public class PrestoS3RecoverableWriterTest {
// ----------------------- S3 general configuration -----------------------
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final long PART_UPLOAD_MIN_SIZE_VALUE = 7L << 20;
private static final int MAX_CONCURRENT_UPLOADS_VALUE = 2;
@@ -53,21 +49,17 @@ public class PrestoS3RecoverableWriterTest {
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final Path basePath = new Path("s3://" + BUCKET + '/' + TEST_DATA_DIR);
-
// ----------------------- Test Lifecycle -----------------------
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
// initialize configuration with valid credentials
final Configuration conf = new Configuration();
- conf.setString("s3.access.key", ACCESS_KEY);
- conf.setString("s3.secret.key", SECRET_KEY);
+ conf.setString("s3.access.key", S3TestCredentials.getS3AccessKey());
+ conf.setString("s3.secret.key", S3TestCredentials.getS3SecretKey());
conf.setLong(PART_UPLOAD_MIN_SIZE, PART_UPLOAD_MIN_SIZE_VALUE);
conf.setInteger(MAX_CONCURRENT_UPLOADS, MAX_CONCURRENT_UPLOADS_VALUE);
@@ -87,7 +79,8 @@ public class PrestoS3RecoverableWriterTest {
@Test(expected = UnsupportedOperationException.class)
public void requestingRecoverableWriterShouldThroughException() throws Exception {
- FlinkS3FileSystem fileSystem = (FlinkS3FileSystem) FileSystem.get(basePath.toUri());
+ URI s3Uri = URI.create(S3TestCredentials.getTestBucketUri());
+ FlinkS3FileSystem fileSystem = (FlinkS3FileSystem) FileSystem.get(s3Uri);
fileSystem.createRecoverableWriter();
}
}
diff --git a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/s3/S3Credentials.java b/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/s3/S3Credentials.java
deleted file mode 100644
index 8f13ffa..0000000
--- a/flink-test-utils-parent/flink-test-utils-junit/src/main/java/org/apache/flink/testutils/s3/S3Credentials.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * 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.testutils.s3;
-
-import org.junit.Assume;
-import org.junit.AssumptionViolatedException;
-
-import javax.annotation.Nullable;
-
-/**
- * Access to credentials to access S3 buckets during integration tests.
- */
-public class S3Credentials {
-
- @Nullable
- private static final String S3_TEST_BUCKET = System.getenv("IT_CASE_S3_BUCKET");
-
- @Nullable
- private static final String S3_TEST_ACCESS_KEY = System.getenv("IT_CASE_S3_ACCESS_KEY");
-
- @Nullable
- private static final String S3_TEST_SECRET_KEY = System.getenv("IT_CASE_S3_SECRET_KEY");
-
- // ------------------------------------------------------------------------
-
- /**
- * Checks whether S3 test credentials are available in the environment variables
- * of this JVM.
- */
- public static boolean credentialsAvailable() {
- return S3_TEST_BUCKET != null && S3_TEST_ACCESS_KEY != null && S3_TEST_SECRET_KEY != null;
- }
-
- /**
- * Checks whether credentials are available in the environment variables of this JVM.
- * If not, throws an {@link AssumptionViolatedException} which causes JUnit tests to be
- * skipped.
- */
- public static void assumeCredentialsAvailable() {
- Assume.assumeTrue("No S3 credentials available in this test's environment", credentialsAvailable());
- }
-
- /**
- * Gets the S3 Access Key.
- *
- * <p>This method throws an exception if the key is not available. Tests should
- * use {@link #assumeCredentialsAvailable()} to skip tests when credentials are not
- * available.
- */
- public String getS3AccessKey() {
- if (S3_TEST_ACCESS_KEY != null) {
- return S3_TEST_ACCESS_KEY;
- }
- else {
- throw new IllegalStateException("S3 test access key not available");
- }
- }
-
- /**
- * Gets the S3 Secret Key.
- *
- * <p>This method throws an exception if the key is not available. Tests should
- * use {@link #assumeCredentialsAvailable()} to skip tests when credentials are not
- * available.
- */
- public String getS3SecretKey() {
- if (S3_TEST_SECRET_KEY != null) {
- return S3_TEST_SECRET_KEY;
- }
- else {
- throw new IllegalStateException("S3 test secret key not available");
- }
- }
-
- /**
- * Gets the URI for the path under which all tests should put their data.
- *
- * <p>This method throws an exception if the bucket was not configured. Tests should
- * use {@link #assumeCredentialsAvailable()} to skip tests when credentials are not
- * available.
- */
- public String getBaseUri() {
- if (S3_TEST_BUCKET != null) {
- return "s3://" + S3_TEST_BUCKET + "/temp/";
- }
- else {
- throw new IllegalStateException("S3 test bucket not available");
- }
- }
-}
diff --git a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTestS3ITCase.java b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTestS3ITCase.java
index e1e95b1..bd9d699 100644
--- a/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTestS3ITCase.java
+++ b/flink-yarn/src/test/java/org/apache/flink/yarn/YarnFileStageTestS3ITCase.java
@@ -23,10 +23,10 @@ import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.fs.hdfs.HadoopFileSystem;
+import org.apache.flink.testutils.s3.S3TestCredentials;
import org.apache.flink.util.TestLogger;
import org.junit.AfterClass;
-import org.junit.Assume;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Rule;
@@ -53,13 +53,8 @@ import static org.junit.Assume.assumeNoException;
*/
public class YarnFileStageTestS3ITCase extends TestLogger {
- private static final String BUCKET = System.getenv("ARTIFACTS_AWS_BUCKET");
-
private static final String TEST_DATA_DIR = "tests-" + UUID.randomUUID();
- private static final String ACCESS_KEY = System.getenv("ARTIFACTS_AWS_ACCESS_KEY");
- private static final String SECRET_KEY = System.getenv("ARTIFACTS_AWS_SECRET_KEY");
-
@ClassRule
public static final TemporaryFolder TEMP_FOLDER = new TemporaryFolder();
@@ -79,9 +74,7 @@ public class YarnFileStageTestS3ITCase extends TestLogger {
@BeforeClass
public static void checkCredentialsAndSetup() throws IOException {
// check whether credentials exist
- Assume.assumeTrue("AWS S3 bucket not configured, skipping test...", BUCKET != null);
- Assume.assumeTrue("AWS S3 access key not configured, skipping test...", ACCESS_KEY != null);
- Assume.assumeTrue("AWS S3 secret key not configured, skipping test...", SECRET_KEY != null);
+ S3TestCredentials.assumeCredentialsAvailable();
skipTest = false;
@@ -115,14 +108,14 @@ public class YarnFileStageTestS3ITCase extends TestLogger {
Map<String /* key */, String /* value */> parameters = new HashMap<>();
// set all different S3 fs implementation variants' configuration keys
- parameters.put("fs.s3a.access.key", ACCESS_KEY);
- parameters.put("fs.s3a.secret.key", SECRET_KEY);
+ parameters.put("fs.s3a.access.key", S3TestCredentials.getS3AccessKey());
+ parameters.put("fs.s3a.secret.key", S3TestCredentials.getS3SecretKey());
- parameters.put("fs.s3.awsAccessKeyId", ACCESS_KEY);
- parameters.put("fs.s3.awsSecretAccessKey", SECRET_KEY);
+ parameters.put("fs.s3.awsAccessKeyId", S3TestCredentials.getS3AccessKey());
+ parameters.put("fs.s3.awsSecretAccessKey", S3TestCredentials.getS3SecretKey());
- parameters.put("fs.s3n.awsAccessKeyId", ACCESS_KEY);
- parameters.put("fs.s3n.awsSecretAccessKey", SECRET_KEY);
+ parameters.put("fs.s3n.awsAccessKeyId", S3TestCredentials.getS3AccessKey());
+ parameters.put("fs.s3n.awsSecretAccessKey", S3TestCredentials.getS3SecretKey());
try (PrintStream out = new PrintStream(new FileOutputStream(hadoopConfig))) {
out.println("<?xml version=\"1.0\"?>");
@@ -155,7 +148,7 @@ public class YarnFileStageTestS3ITCase extends TestLogger {
private void testRecursiveUploadForYarn(String scheme, String pathSuffix) throws Exception {
++numRecursiveUploadTests;
- final Path basePath = new Path(scheme + "://" + BUCKET + '/' + TEST_DATA_DIR);
+ final Path basePath = new Path(S3TestCredentials.getTestBucketUriWithScheme(scheme) + TEST_DATA_DIR);
final HadoopFileSystem fs = (HadoopFileSystem) basePath.getFileSystem();
assumeFalse(fs.exists(basePath));