You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by sh...@apache.org on 2023/03/03 07:39:42 UTC
[flink] branch release-1.17 updated (7dd61c31714 -> 86e12eb3fce)
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a change to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
from 7dd61c31714 [FLINK-31288][doc] Update doc ralated to overdraft buffer.
new 594010624f8 [FLINK-31092][sql-gateway] Fix OperationManager can not kill the running task by force
new 86e12eb3fce [FLINK-31092][table-common] Fix ServiceLoaderUtil keeps loading even though the classloader has been closed
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/table/gateway/api/utils/ThreadUtils.java | 4 +-
.../service/operation/OperationManager.java | 47 +++++++++++-
.../service/operation/OperationManagerTest.java | 29 ++++++++
.../apache/flink/table/factories/FactoryUtil.java | 52 ++++++++------
.../flink/table/factories/ServiceLoaderUtil.java | 83 ----------------------
.../flink/table/factories/FactoryUtilTest.java | 20 ++++++
6 files changed, 128 insertions(+), 107 deletions(-)
delete mode 100644 flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/ServiceLoaderUtil.java
[flink] 02/02: [FLINK-31092][table-common] Fix ServiceLoaderUtil keeps loading even though the classloader has been closed
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 86e12eb3fcec54d234154e59f8cb0557fc616494
Author: Shengkai <10...@qq.com>
AuthorDate: Thu Mar 2 20:05:51 2023 +0800
[FLINK-31092][table-common] Fix ServiceLoaderUtil keeps loading even though the classloader has been closed
This closes #22072
Co-authored-by: Matthias Pohl <ma...@aiven.io>
---
.../apache/flink/table/factories/FactoryUtil.java | 52 ++++++++------
.../flink/table/factories/ServiceLoaderUtil.java | 83 ----------------------
.../flink/table/factories/FactoryUtilTest.java | 20 ++++++
3 files changed, 50 insertions(+), 105 deletions(-)
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
index d90dbc6c2f8..e22ba4051da 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/FactoryUtil.java
@@ -52,11 +52,12 @@ import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
-import java.util.LinkedList;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
+import java.util.ServiceLoader;
import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -801,27 +802,34 @@ public final class FactoryUtil {
}
static List<Factory> discoverFactories(ClassLoader classLoader) {
- final List<Factory> result = new LinkedList<>();
- ServiceLoaderUtil.load(Factory.class, classLoader)
- .forEach(
- loadResult -> {
- if (loadResult.hasFailed()) {
- if (loadResult.getError() instanceof NoClassDefFoundError) {
- LOG.debug(
- "NoClassDefFoundError when loading a "
- + Factory.class
- + ". This is expected when trying to load a format dependency but no flink-connector-files is loaded.",
- loadResult.getError());
- // After logging, we just ignore this failure
- return;
- }
- throw new TableException(
- "Unexpected error when trying to load service provider for factories.",
- loadResult.getError());
- }
- result.add(loadResult.getService());
- });
- return result;
+ final Iterator<Factory> serviceLoaderIterator =
+ ServiceLoader.load(Factory.class, classLoader).iterator();
+
+ final List<Factory> loadResults = new ArrayList<>();
+ while (true) {
+ try {
+ // error handling should also be applied to the hasNext() call because service
+ // loading might cause problems here as well
+ if (!serviceLoaderIterator.hasNext()) {
+ break;
+ }
+
+ loadResults.add(serviceLoaderIterator.next());
+ } catch (Throwable t) {
+ if (t instanceof NoClassDefFoundError) {
+ LOG.debug(
+ "NoClassDefFoundError when loading a "
+ + Factory.class.getCanonicalName()
+ + ". This is expected when trying to load a format dependency but no flink-connector-files is loaded.",
+ t);
+ } else {
+ throw new TableException(
+ "Unexpected error when trying to load service provider.", t);
+ }
+ }
+ }
+
+ return loadResults;
}
private static String stringifyOption(String key, String value) {
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/ServiceLoaderUtil.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/ServiceLoaderUtil.java
deleted file mode 100644
index 620e9c3230a..00000000000
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/factories/ServiceLoaderUtil.java
+++ /dev/null
@@ -1,83 +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.table.factories;
-
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-import java.util.NoSuchElementException;
-import java.util.ServiceLoader;
-
-/** This class contains utilities to deal with {@link ServiceLoader}. */
-class ServiceLoaderUtil {
-
- /**
- * This method behaves similarly to {@link ServiceLoader#load(Class, ClassLoader)}, but it
- * returns a list with the results of the iteration, wrapping the iteration failures such as
- * {@link NoClassDefFoundError}.
- */
- static <T> List<LoadResult<T>> load(Class<T> clazz, ClassLoader classLoader) {
- List<LoadResult<T>> loadResults = new ArrayList<>();
-
- Iterator<T> serviceLoaderIterator = ServiceLoader.load(clazz, classLoader).iterator();
-
- while (true) {
- try {
- T next = serviceLoaderIterator.next();
- loadResults.add(new LoadResult<>(next));
- } catch (NoSuchElementException e) {
- break;
- } catch (Throwable t) {
- loadResults.add(new LoadResult<>(t));
- }
- }
-
- return loadResults;
- }
-
- static class LoadResult<T> {
- private final T service;
- private final Throwable error;
-
- private LoadResult(T service, Throwable error) {
- this.service = service;
- this.error = error;
- }
-
- private LoadResult(T service) {
- this(service, null);
- }
-
- private LoadResult(Throwable error) {
- this(null, error);
- }
-
- public boolean hasFailed() {
- return error != null;
- }
-
- public Throwable getError() {
- return error;
- }
-
- public T getService() {
- return service;
- }
- }
-}
diff --git a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java
index 45ef9bda2ec..6f6d2a8f40c 100644
--- a/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java
+++ b/flink-table/flink-table-common/src/test/java/org/apache/flink/table/factories/FactoryUtilTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.table.factories;
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.ConfigOptions;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.table.api.ValidationException;
import org.apache.flink.table.catalog.Catalog;
import org.apache.flink.table.catalog.CommonCatalogOptions;
@@ -33,12 +34,15 @@ import org.apache.flink.table.factories.TestFormatFactory.DecodingFormatMock;
import org.apache.flink.table.factories.TestFormatFactory.EncodingFormatMock;
import org.apache.flink.table.factories.utils.FactoryMocks;
import org.apache.flink.testutils.ClassLoaderUtils;
+import org.apache.flink.util.FlinkUserCodeClassLoaders;
+import org.apache.flink.util.MutableURLClassLoader;
import org.assertj.core.api.AbstractThrowableAssert;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import java.io.IOException;
+import java.net.URL;
import java.net.URLClassLoader;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -668,6 +672,22 @@ class FactoryUtilTest {
.contains(serializationSchemaImplementationName);
}
+ @Test
+ void testDiscoverFactoryFromClosedClassLoader() throws Exception {
+ MutableURLClassLoader classLoader =
+ FlinkUserCodeClassLoaders.create(
+ new URL[0], FactoryUtilTest.class.getClassLoader(), new Configuration());
+ classLoader.close();
+ assertThatThrownBy(() -> FactoryUtil.discoverFactory(classLoader, Factory.class, "test"))
+ .satisfies(
+ FlinkAssertions.anyCauseMatches(
+ IllegalStateException.class,
+ "Trying to access closed classloader. Please check if you store classloaders directly "
+ + "or indirectly in static fields. If the stacktrace suggests that the leak occurs in a third "
+ + "party library and cannot be fixed immediately, you can disable this check with the "
+ + "configuration 'classloader.check-leaked-classloader'"));
+ }
+
// --------------------------------------------------------------------------------------------
// Helper methods
// --------------------------------------------------------------------------------------------
[flink] 01/02: [FLINK-31092][sql-gateway] Fix OperationManager can not kill the running task by force
Posted by sh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
shengkai pushed a commit to branch release-1.17
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 594010624f8084efd99d6d405b5310ab24013aeb
Author: Shengkai <10...@qq.com>
AuthorDate: Thu Mar 2 17:57:55 2023 +0800
[FLINK-31092][sql-gateway] Fix OperationManager can not kill the running task by force
---
.../flink/table/gateway/api/utils/ThreadUtils.java | 4 +-
.../service/operation/OperationManager.java | 47 +++++++++++++++++++++-
.../service/operation/OperationManagerTest.java | 29 +++++++++++++
3 files changed, 78 insertions(+), 2 deletions(-)
diff --git a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/utils/ThreadUtils.java b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/utils/ThreadUtils.java
index 330d8ed9eb9..9408c61ecbf 100644
--- a/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/utils/ThreadUtils.java
+++ b/flink-table/flink-sql-gateway-api/src/main/java/org/apache/flink/table/gateway/api/utils/ThreadUtils.java
@@ -19,6 +19,7 @@
package org.apache.flink.table.gateway.api.utils;
import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.concurrent.ExecutorThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -47,6 +48,7 @@ public class ThreadUtils {
poolQueueSize,
keepAliveMs,
TimeUnit.MILLISECONDS,
- new SynchronousQueue<>());
+ new SynchronousQueue<>(),
+ new ExecutorThreadFactory(threadPoolName));
}
}
diff --git a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java
index 4e16cb5dd6c..8b239abc771 100644
--- a/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java
+++ b/flink-table/flink-sql-gateway/src/main/java/org/apache/flink/table/gateway/service/operation/OperationManager.java
@@ -20,6 +20,7 @@ package org.apache.flink.table.gateway.service.operation;
import org.apache.flink.annotation.Internal;
import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.gateway.api.operation.OperationHandle;
import org.apache.flink.table.gateway.api.operation.OperationStatus;
@@ -34,8 +35,11 @@ import org.apache.flink.table.gateway.service.utils.SqlExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.lang.reflect.Field;
+import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.FutureTask;
@@ -366,7 +370,6 @@ public class OperationManager {
String.format(
"Failed to convert the Operation Status from %s to %s for %s.",
currentStatus, toStatus, operationHandle);
- LOG.error(message);
throw new SqlGatewayException(message);
}
} while (!status.compareAndSet(currentStatus, toStatus));
@@ -384,6 +387,7 @@ public class OperationManager {
private void closeResources() {
if (invocation != null && !invocation.isDone()) {
invocation.cancel(true);
+ stopExecutionByForce(invocation);
LOG.debug(String.format("Cancel the operation %s.", operationHandle));
}
@@ -400,6 +404,47 @@ public class OperationManager {
// when status is error.
updateState(OperationStatus.ERROR);
}
+
+ private void stopExecutionByForce(FutureTask<?> invocation) {
+ // thread is cleaned async, waiting for a while
+ Deadline deadline = Deadline.fromNow(Duration.ofSeconds(1));
+ while (deadline.hasTimeLeft()) {
+ Optional<Thread> threadOptional = getThreadInFuture(invocation);
+ if (!threadOptional.isPresent()) {
+ // thread has been cleaned up
+ return;
+ }
+ }
+ Optional<Thread> threadOptional = getThreadInFuture(invocation);
+ if (threadOptional.isPresent()) {
+ // we have to use Thread.stop() here, because this can
+ // guarantee thread to be stopped, even there is some
+ // potential consistent problem, we are fine with it.
+ Thread thread = threadOptional.get();
+ LOG.info(
+ "\"Future.cancel(true)\" can't cleanup current thread {}, using \"Thread.stop()\" instead.",
+ thread.getName());
+ try {
+ thread.stop();
+ } catch (Throwable e) {
+ // catch all errors to project the sqlserver
+ LOG.error("Failed to stop thread: " + thread.getName(), e);
+ }
+ }
+ }
+
+ private Optional<Thread> getThreadInFuture(FutureTask<?> invocation) {
+ try {
+ Class<?> k = FutureTask.class;
+ Field runnerField = k.getDeclaredField("runner");
+ runnerField.setAccessible(true);
+ Thread t = (Thread) runnerField.get(invocation);
+ return Optional.of(t);
+ } catch (Throwable e) {
+ // can't get thread
+ return Optional.empty();
+ }
+ }
}
// -------------------------------------------------------------------------------------------
diff --git a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/operation/OperationManagerTest.java b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/operation/OperationManagerTest.java
index b1a1863ef82..e4d95f1bae1 100644
--- a/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/operation/OperationManagerTest.java
+++ b/flink-table/flink-sql-gateway/src/test/java/org/apache/flink/table/gateway/service/operation/OperationManagerTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.gateway.service.operation;
+import org.apache.flink.core.testutils.CommonTestUtils;
import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.ResultKind;
@@ -38,10 +39,12 @@ import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
+import java.time.Duration;
import java.util.Collections;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.flink.table.api.internal.StaticResultProvider.SIMPLE_ROW_DATA_TO_STRING_CONVERTER;
import static org.apache.flink.table.gateway.api.results.ResultSet.ResultType.PAYLOAD;
@@ -125,6 +128,32 @@ public class OperationManagerTest {
.isEqualTo(OperationStatus.CANCELED);
}
+ @Test
+ public void testCancelOperationByForce() throws Exception {
+ AtomicReference<Throwable> exception = new AtomicReference<>(null);
+ OperationHandle operationHandle =
+ operationManager.submitOperation(
+ () -> {
+ try {
+ // mock cpu busy task that doesn't interrupt system call
+ while (true) {}
+ } catch (Throwable t) {
+ exception.set(t);
+ throw t;
+ }
+ });
+
+ threadFactory.newThread(() -> operationManager.cancelOperation(operationHandle)).start();
+ operationManager.awaitOperationTermination(operationHandle);
+
+ assertThat(operationManager.getOperationInfo(operationHandle).getStatus())
+ .isEqualTo(OperationStatus.CANCELED);
+ CommonTestUtils.waitUtil(
+ () -> exception.get() != null,
+ Duration.ofSeconds(10),
+ "Failed to kill the task with infinite loop.");
+ }
+
@Test
public void testCloseOperation() throws Exception {
CountDownLatch endRunningLatch = new CountDownLatch(1);