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 2020/05/19 13:38:08 UTC
[flink] branch release-1.11 updated (14f0bce -> fbe388f)
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 14f0bce [FLINK-16383][task] Do not relay notifyCheckpointComplete to closed operators.
new fa0362f [FLINK-17634][rest] Reject multiple registration for the same endpoint
new fbe388f [FLINK-16970][dist] Bundle JMXReporter separately from dist jar
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
flink-dist/pom.xml | 27 +++++++++++---
.../src/main/assemblies/plugins.xml | 35 +++++++++++-------
.../flink/runtime/rest/RestServerEndpoint.java | 43 +++++++++++++++++-----
.../runtime/rest/RestServerEndpointITCase.java | 21 ++++++++++-
4 files changed, 96 insertions(+), 30 deletions(-)
copy flink-scala-shell/src/test/assembly/test-scalashell-customjar-assembly.xml => flink-dist/src/main/assemblies/plugins.xml (52%)
[flink] 02/02: [FLINK-16970][dist] Bundle JMXReporter separately
from dist jar
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fbe388fca460979aa049a904c64201ec42b7cd1f
Author: molsionmo <52...@users.noreply.github.com>
AuthorDate: Tue May 19 21:27:11 2020 +0800
[FLINK-16970][dist] Bundle JMXReporter separately from dist jar
---
flink-dist/pom.xml | 27 ++++++++++++++-----
flink-dist/src/main/assemblies/plugins.xml | 42 ++++++++++++++++++++++++++++++
2 files changed, 63 insertions(+), 6 deletions(-)
diff --git a/flink-dist/pom.xml b/flink-dist/pom.xml
index 87718be..558ad57 100644
--- a/flink-dist/pom.xml
+++ b/flink-dist/pom.xml
@@ -100,12 +100,6 @@ under the License.
<artifactId>flink-metrics-core</artifactId>
<version>${project.version}</version>
</dependency>
-
- <dependency>
- <groupId>org.apache.flink</groupId>
- <artifactId>flink-metrics-jmx_${scala.binary.version}</artifactId>
- <version>${project.version}</version>
- </dependency>
<dependency>
<groupId>org.apache.flink</groupId>
@@ -280,6 +274,13 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-metrics-jmx_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-metrics-statsd</artifactId>
<version>${project.version}</version>
<scope>provided</scope>
@@ -793,6 +794,20 @@ under the License.
<appendAssemblyId>false</appendAssemblyId>
</configuration>
</execution>
+ <execution>
+ <id>plugins</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ <configuration>
+ <descriptors>
+ <descriptor>src/main/assemblies/plugins.xml</descriptor>
+ </descriptors>
+ <finalName>flink-${project.version}-bin</finalName>
+ <appendAssemblyId>false</appendAssemblyId>
+ </configuration>
+ </execution>
</executions>
</plugin>
diff --git a/flink-dist/src/main/assemblies/plugins.xml b/flink-dist/src/main/assemblies/plugins.xml
new file mode 100644
index 0000000..7c0baed
--- /dev/null
+++ b/flink-dist/src/main/assemblies/plugins.xml
@@ -0,0 +1,42 @@
+<!--
+ ~ 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.
+ -->
+
+<assembly
+ xmlns="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/plugins/maven-assembly-plugin/assembly/1.1.0 http://maven.apache.org/xsd/assembly-1.1.0.xsd">
+ <id>plugins</id>
+ <formats>
+ <format>dir</format>
+ </formats>
+
+ <includeBaseDirectory>true</includeBaseDirectory>
+ <baseDirectory>flink-${project.version}</baseDirectory>
+
+ <files>
+ <!-- Metrics -->
+
+ <file>
+ <source>../flink-metrics/flink-metrics-jmx/target/flink-metrics-jmx_${scala.binary.version}-${project.version}.jar</source>
+ <outputDirectory>plugins/metrics_jmx/</outputDirectory>
+ <destName>flink-metrics-jmx-${project.version}.jar</destName>
+ <fileMode>0644</fileMode>
+ </file>
+ </files>
+
+</assembly>
[flink] 01/02: [FLINK-17634][rest] Reject multiple registration for
the same endpoint
Posted by ch...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit fa0362f430165f6ba9e25a50ce5d98e793b6ab38
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Tue May 12 14:56:29 2020 +0200
[FLINK-17634][rest] Reject multiple registration for the same endpoint
---
.../flink/runtime/rest/RestServerEndpoint.java | 43 +++++++++++++++++-----
.../runtime/rest/RestServerEndpointITCase.java | 21 ++++++++++-
2 files changed, 54 insertions(+), 10 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
index a264037..f575da2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/RestServerEndpoint.java
@@ -64,6 +64,7 @@ import java.nio.file.Path;
import java.util.Arrays;
import java.util.Collections;
import java.util.Comparator;
+import java.util.HashSet;
import java.util.IdentityHashMap;
import java.util.Iterator;
import java.util.List;
@@ -140,7 +141,6 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
final Router router = new Router();
final CompletableFuture<String> restAddressFuture = new CompletableFuture<>();
- final Set<ChannelInboundHandler> distinctHandlers = Collections.newSetFromMap(new IdentityHashMap<>());
handlers = initializeHandlers(restAddressFuture);
@@ -155,14 +155,8 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
handlers,
RestHandlerUrlComparator.INSTANCE);
- handlers.forEach(handler -> {
- if (distinctHandlers.contains(handler.f1)) {
- throw new FlinkRuntimeException("Duplicate REST handler instance found."
- + " Please ensure each instance is registered only once.");
- }
- registerHandler(router, handler, log);
- distinctHandlers.add(handler.f1);
- });
+ checkAllEndpointsAndHandlersAreUnique(handlers);
+ handlers.forEach(handler -> registerHandler(router, handler, log));
ChannelInitializer<SocketChannel> initializer = new ChannelInitializer<SocketChannel>() {
@@ -491,6 +485,37 @@ public abstract class RestServerEndpoint implements AutoCloseableAsync {
}
}
+ private static void checkAllEndpointsAndHandlersAreUnique(final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers) {
+ // check for all handlers that
+ // 1) the instance is only registered once
+ // 2) only 1 handler is registered for each endpoint (defined by (version, method, url))
+ // technically the first check is redundant since a duplicate instance also returns the same headers which
+ // should fail the second check, but we get a better error message
+ final Set<String> uniqueEndpoints = new HashSet<>();
+ final Set<ChannelInboundHandler> distinctHandlers = Collections.newSetFromMap(new IdentityHashMap<>());
+ for (Tuple2<RestHandlerSpecification, ChannelInboundHandler> handler : handlers) {
+ boolean isNewHandler = distinctHandlers.add(handler.f1);
+ if (!isNewHandler) {
+ throw new FlinkRuntimeException("Duplicate REST handler instance found."
+ + " Please ensure each instance is registered only once.");
+ }
+
+ final RestHandlerSpecification headers = handler.f0;
+ for (RestAPIVersion supportedAPIVersion : headers.getSupportedAPIVersions()) {
+ final String parameterizedEndpoint = supportedAPIVersion.toString() + headers.getHttpMethod() + headers.getTargetRestEndpointURL();
+ // normalize path parameters; distinct path parameters still clash at runtime
+ final String normalizedEndpoint = parameterizedEndpoint.replaceAll(":[\\w-]+", ":param");
+ boolean isNewEndpoint = uniqueEndpoints.add(normalizedEndpoint);
+ if (!isNewEndpoint) {
+ throw new FlinkRuntimeException(
+ String.format(
+ "REST handler registration overlaps with another registration for: version=%s, method=%s, url=%s.",
+ supportedAPIVersion, headers.getHttpMethod(), headers.getTargetRestEndpointURL()));
+ }
+ }
+ }
+ }
+
/**
* Comparator for Rest URLs.
*
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
index 0a95a3c..c941c45 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestServerEndpointITCase.java
@@ -595,12 +595,31 @@ public class RestServerEndpointITCase extends TestLogger {
}
@Test
+ public void testEndpointsMustBeUnique() throws Exception {
+ final RestServerEndpointConfiguration serverConfig = RestServerEndpointConfiguration.fromConfiguration(config);
+
+ final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = Arrays.asList(
+ Tuple2.of(new TestHeaders(), testHandler),
+ Tuple2.of(new TestHeaders(), testUploadHandler)
+ );
+
+ assertThrows("REST handler registration",
+ FlinkRuntimeException.class,
+ () -> {
+ try (TestRestServerEndpoint restServerEndpoint = new TestRestServerEndpoint(serverConfig, handlers)) {
+ restServerEndpoint.start();
+ return null;
+ }
+ });
+ }
+
+ @Test
public void testDuplicateHandlerRegistrationIsForbidden() throws Exception {
final RestServerEndpointConfiguration serverConfig = RestServerEndpointConfiguration.fromConfiguration(config);
final List<Tuple2<RestHandlerSpecification, ChannelInboundHandler>> handlers = Arrays.asList(
Tuple2.of(new TestHeaders(), testHandler),
- Tuple2.of(new TestHeaders(), testHandler)
+ Tuple2.of(TestUploadHeaders.INSTANCE, testHandler)
);
assertThrows("Duplicate REST handler",