You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/11/05 08:49:19 UTC
[flink] 02/04: [FLINK-24603][e2e] Allow arbitrary jars to be added
to the distribution
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 24fd6d551f776f229acc89f3198cbcf8d8966ea6
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Thu Oct 28 12:19:40 2021 +0200
[FLINK-24603][e2e] Allow arbitrary jars to be added to the distribution
Required for adding a user-provided Scala library to lib.
---
.../flink/tests/util/flink/FlinkDistribution.java | 17 +++++++++
.../flink/tests/util/flink/FlinkResourceSetup.java | 21 ++++++++++--
.../apache/flink/tests/util/flink/JarAddition.java | 40 ++++++++++++++++++++++
.../util/flink/LocalStandaloneFlinkResource.java | 3 ++
4 files changed, 79 insertions(+), 2 deletions(-)
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
index 4f779d5..cd3d935 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkDistribution.java
@@ -33,6 +33,7 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMap
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
+import org.apache.commons.io.FilenameUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.event.Level;
@@ -213,6 +214,22 @@ final class FlinkDistribution {
.runBlocking(timeout);
}
+ public void performJarAddition(JarAddition addition) throws IOException {
+ final Path target = mapJarLocationToPath(addition.getTarget());
+ final Path sourceJar = addition.getJar();
+
+ final String jarNameWithoutExtension =
+ FilenameUtils.removeExtension(sourceJar.getFileName().toString());
+
+ // put the jar into a directory within the target location; this is primarily needed for
+ // plugins/, but also works for lib/
+ final Path targetJar =
+ target.resolve(jarNameWithoutExtension).resolve(sourceJar.getFileName());
+ Files.createDirectories(targetJar.getParent());
+
+ Files.copy(sourceJar, targetJar);
+ }
+
public void performJarOperation(JarOperation operation) throws IOException {
final Path source = mapJarLocationToPath(operation.getSource());
final Path target = mapJarLocationToPath(operation.getTarget());
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceSetup.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceSetup.java
index 7c8a1a1..1256e09 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceSetup.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/FlinkResourceSetup.java
@@ -22,6 +22,7 @@ import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
+import java.nio.file.Path;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -32,11 +33,15 @@ public class FlinkResourceSetup {
@Nullable private final Configuration config;
private final Collection<JarOperation> jarOperations;
+ private final Collection<JarAddition> jarAdditions;
private FlinkResourceSetup(
- @Nullable Configuration config, Collection<JarOperation> jarOperations) {
+ @Nullable Configuration config,
+ Collection<JarOperation> jarOperations,
+ Collection<JarAddition> jarAdditions) {
this.config = config;
this.jarOperations = Preconditions.checkNotNull(jarOperations);
+ this.jarAdditions = Preconditions.checkNotNull(jarAdditions);
}
public Optional<Configuration> getConfig() {
@@ -47,6 +52,10 @@ public class FlinkResourceSetup {
return jarOperations;
}
+ public Collection<JarAddition> getJarAdditions() {
+ return jarAdditions;
+ }
+
public static FlinkResourceSetupBuilder builder() {
return new FlinkResourceSetupBuilder();
}
@@ -56,6 +65,7 @@ public class FlinkResourceSetup {
private Configuration config;
private final Collection<JarOperation> jarOperations = new ArrayList<>();
+ private final Collection<JarAddition> jarAdditions = new ArrayList<>();
private FlinkResourceSetupBuilder() {}
@@ -64,6 +74,11 @@ public class FlinkResourceSetup {
return this;
}
+ public FlinkResourceSetupBuilder addJar(Path jar, JarLocation target) {
+ this.jarAdditions.add(new JarAddition(jar, target));
+ return this;
+ }
+
public FlinkResourceSetupBuilder moveJar(
String jarNamePrefix, JarLocation source, JarLocation target) {
this.jarOperations.add(
@@ -82,7 +97,9 @@ public class FlinkResourceSetup {
public FlinkResourceSetup build() {
return new FlinkResourceSetup(
- config, Collections.unmodifiableCollection(jarOperations));
+ config,
+ Collections.unmodifiableCollection(jarOperations),
+ Collections.unmodifiableCollection(jarAdditions));
}
}
}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/JarAddition.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/JarAddition.java
new file mode 100644
index 0000000..8ea42b7
--- /dev/null
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/JarAddition.java
@@ -0,0 +1,40 @@
+/*
+ * 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.tests.util.flink;
+
+import java.nio.file.Path;
+
+/** Represents an add operation for a jar. */
+class JarAddition {
+
+ private final Path jar;
+ private final JarLocation target;
+
+ JarAddition(Path jar, JarLocation target) {
+ this.jar = jar;
+ this.target = target;
+ }
+
+ public Path getJar() {
+ return jar;
+ }
+
+ public JarLocation getTarget() {
+ return target;
+ }
+}
diff --git a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/LocalStandaloneFlinkResource.java b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/LocalStandaloneFlinkResource.java
index 429e4f5..b5e2d75 100644
--- a/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/LocalStandaloneFlinkResource.java
+++ b/flink-end-to-end-tests/flink-end-to-end-tests-common/src/main/java/org/apache/flink/tests/util/flink/LocalStandaloneFlinkResource.java
@@ -85,6 +85,9 @@ public class LocalStandaloneFlinkResource implements FlinkResource {
for (JarOperation jarOperation : setup.getJarOperations()) {
distribution.performJarOperation(jarOperation);
}
+ for (JarAddition jarAddition : setup.getJarAdditions()) {
+ distribution.performJarAddition(jarAddition);
+ }
if (setup.getConfig().isPresent()) {
distribution.appendConfiguration(setup.getConfig().get());
}