You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2021/02/11 17:50:57 UTC
[flink] branch release-1.11 updated (3ea8653 -> 30351d6)
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a change to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git.
from 3ea8653 [FLINK-21312][checkpointing] Unset IsStoppingBySyncSavepoint from the LegacySourceThread
new d69d134 [FLINK-21138] - User ClassLoader in KvStateServerHandler
new 5f6a804 [FLINK-21138][tests] Register serializer instance in AbstractQueryableStateTestBase.testCustomKryoSerializerHandling
new 95281a7 [FLINK-21138] Fix classloader on QueryableStateClient
new 30351d6 [FLINK-21138] Explicit Classloader in QueryableStateClient
The 4 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:
.../client/QueryableStateClient.java | 26 +++++-
.../flink-queryable-state-runtime/pom.xml | 8 ++
.../server/KvStateServerHandler.java | 20 ++--
.../itcases/AbstractQueryableStateTestBase.java | 103 +++++++++++++++++++++
.../flink/queryablestate/network/ClientTest.java | 3 +-
.../network/KvStateServerHandlerTest.java | 3 +-
.../apache/flink/runtime/query/KvStateEntry.java | 9 +-
.../flink/runtime/query/KvStateRegistry.java | 6 +-
.../flink/runtime/query/TaskKvStateRegistry.java | 10 +-
.../runtime/state/AbstractKeyedStateBackend.java | 2 +-
.../flink/runtime/query/KvStateRegistryTest.java | 24 ++++-
11 files changed, 194 insertions(+), 20 deletions(-)
[flink] 01/04: [FLINK-21138] - User ClassLoader in
KvStateServerHandler
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d69d134590f46fea74ac1f7c664435393625533f
Author: Maciek Próchniak <mp...@touk.pl>
AuthorDate: Mon Feb 8 13:11:00 2021 +0100
[FLINK-21138] - User ClassLoader in KvStateServerHandler
---
.../flink-queryable-state-runtime/pom.xml | 8 ++
.../server/KvStateServerHandler.java | 20 +++--
.../itcases/AbstractQueryableStateTestBase.java | 92 ++++++++++++++++++++++
.../flink/queryablestate/network/ClientTest.java | 3 +-
.../network/KvStateServerHandlerTest.java | 3 +-
.../apache/flink/runtime/query/KvStateEntry.java | 9 ++-
.../flink/runtime/query/KvStateRegistry.java | 6 +-
.../flink/runtime/query/TaskKvStateRegistry.java | 10 ++-
.../runtime/state/AbstractKeyedStateBackend.java | 2 +-
.../flink/runtime/query/KvStateRegistryTest.java | 24 +++++-
10 files changed, 158 insertions(+), 19 deletions(-)
diff --git a/flink-queryable-state/flink-queryable-state-runtime/pom.xml b/flink-queryable-state/flink-queryable-state-runtime/pom.xml
index ca0ccf2..63fae9c 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/pom.xml
+++ b/flink-queryable-state/flink-queryable-state-runtime/pom.xml
@@ -65,6 +65,14 @@ under the License.
<dependency>
<groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flink</groupId>
<artifactId>flink-statebackend-rocksdb_${scala.binary.version}</artifactId>
<version>${project.version}</version>
<scope>test</scope>
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
index 0ffb468..798008e 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/main/java/org/apache/flink/queryablestate/server/KvStateServerHandler.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.query.KvStateInfo;
import org.apache.flink.runtime.query.KvStateRegistry;
import org.apache.flink.runtime.state.internal.InternalKvState;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.LambdaUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandler;
@@ -109,14 +110,19 @@ public class KvStateServerHandler
final KvStateEntry<K, N, V> entry, final byte[] serializedKeyAndNamespace)
throws Exception {
- final InternalKvState<K, N, V> state = entry.getState();
- final KvStateInfo<K, N, V> infoForCurrentThread = entry.getInfoForCurrentThread();
+ return LambdaUtil.withContextClassLoader(
+ entry.getUserClassLoader(),
+ () -> {
+ final InternalKvState<K, N, V> state = entry.getState();
+ final KvStateInfo<K, N, V> infoForCurrentThread =
+ entry.getInfoForCurrentThread();
- return state.getSerializedValue(
- serializedKeyAndNamespace,
- infoForCurrentThread.getKeySerializer(),
- infoForCurrentThread.getNamespaceSerializer(),
- infoForCurrentThread.getStateValueSerializer());
+ return state.getSerializedValue(
+ serializedKeyAndNamespace,
+ infoForCurrentThread.getKeySerializer(),
+ infoForCurrentThread.getNamespaceSerializer(),
+ infoForCurrentThread.getStateValueSerializer());
+ });
}
@Override
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
index 9b7c11e..2c0ca6c 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
@@ -45,6 +45,7 @@ import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.client.ClientUtils;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.client.program.ProgramInvocationException;
@@ -68,18 +69,26 @@ import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunctio
import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
import org.apache.flink.streaming.api.operators.OneInputStreamOperator;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.testutils.ClassLoaderUtils;
import org.apache.flink.util.Collector;
import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.LambdaUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
+import com.esotericsoftware.kryo.Serializer;
import org.junit.Assert;
import org.junit.Before;
+import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import java.io.IOException;
+import java.net.URLClassLoader;
import java.time.Duration;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -121,6 +130,8 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
protected static int maxParallelism;
+ @ClassRule public static TemporaryFolder classloaderFolder = new TemporaryFolder();
+
@Before
public void setUp() throws Exception {
// NOTE: do not use a shared instance for all tests as the tests may break
@@ -362,6 +373,65 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
}
}
+ /** This test checks if custom Kryo serializers are loaded with correct classloader. */
+ @Test
+ public void testCustomKryoSerializerHandling() throws Exception {
+ final Deadline deadline = Deadline.now().plus(TEST_TIMEOUT);
+ final long numElements = 1;
+ final String stateName = "teriberka";
+
+ final String customSerializerClassName = "CustomKryo";
+ final URLClassLoader userClassLoader =
+ createLoaderWithCustomKryoSerializer(customSerializerClassName);
+
+ StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
+ env.setStateBackend(stateBackend);
+ env.setParallelism(maxParallelism);
+ // Very important, because cluster is shared between tests and we
+ // don't explicitly check that all slots are available before
+ // submitting.
+ env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000L));
+
+ // Custom serializer is not needed, it's used just to check if serialization works.
+ @SuppressWarnings("unchecked")
+ Class<Serializer<?>> customSerializerClass =
+ (Class<Serializer<?>>) userClassLoader.loadClass(customSerializerClassName);
+ env.getConfig().addDefaultKryoSerializer(Byte.class, customSerializerClass);
+
+ // Here we *force* using Kryo, to check if custom serializers are handled correctly WRT
+ // classloading
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ ValueStateDescriptor<Tuple2<Integer, Long>> valueState =
+ new ValueStateDescriptor<>("any", new GenericTypeInfo(Tuple2.class));
+
+ env.addSource(new TestAscendingValueSource(numElements))
+ .keyBy(
+ new KeySelector<Tuple2<Integer, Long>, Integer>() {
+ private static final long serialVersionUID = 7662520075515707428L;
+
+ @Override
+ public Integer getKey(Tuple2<Integer, Long> value) {
+ return value.f0;
+ }
+ })
+ .asQueryableState(stateName, valueState);
+
+ try (AutoCancellableJob autoCancellableJob =
+ new AutoCancellableJob(deadline, clusterClient, env)) {
+
+ final JobID jobId = autoCancellableJob.getJobId();
+ final JobGraph jobGraph = autoCancellableJob.getJobGraph();
+ jobGraph.setClasspaths(Arrays.asList(userClassLoader.getURLs()));
+
+ clusterClient.submitJob(jobGraph).get();
+ LambdaUtil.withContextClassLoader(
+ userClassLoader,
+ () ->
+ executeValueQuery(
+ deadline, client, jobId, stateName, valueState, numElements));
+ }
+ }
+
/**
* Tests that the correct exception is thrown if the query contains a wrong jobId or wrong
* queryable state name.
@@ -1478,4 +1548,26 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
assertTrue("Did not succeed query", success);
}
}
+
+ private static URLClassLoader createLoaderWithCustomKryoSerializer(String className)
+ throws IOException {
+ return ClassLoaderUtils.compileAndLoadJava(
+ classloaderFolder.newFolder(),
+ className + ".java",
+ "import com.esotericsoftware.kryo.Kryo;\n"
+ + "import com.esotericsoftware.kryo.Serializer;\n"
+ + "import com.esotericsoftware.kryo.io.Input;\n"
+ + "import com.esotericsoftware.kryo.io.Output;\n"
+ + "import java.io.Serializable;\n"
+ + "public class "
+ + className
+ + " extends Serializer<Byte> implements Serializable {\n"
+ + " @Override\n"
+ + " public void write(Kryo kryo, Output output, Byte testJob) {}\n"
+ + " @Override\n"
+ + " public Byte read(Kryo kryo, Input input, Class<Byte> aClass) {\n"
+ + " return null;\n"
+ + " }\n"
+ + "}\n");
+ }
}
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java
index 6e888cb..e89a14f 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java
@@ -741,7 +741,8 @@ public class ClientTest extends TestLogger {
new JobVertexID(),
new KeyGroupRange(0, 0),
"any",
- kvState);
+ kvState,
+ getClass().getClassLoader());
}
final Client<KvStateInternalRequest, KvStateResponse> finalClient = client;
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
index f156d9a..6df22a8 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/KvStateServerHandlerTest.java
@@ -383,7 +383,8 @@ public class KvStateServerHandlerTest extends TestLogger {
new JobVertexID(),
new KeyGroupRange(0, 0),
"vanilla",
- kvState);
+ kvState,
+ getClass().getClassLoader());
KvStateInternalRequest request = new KvStateInternalRequest(kvStateId, new byte[0]);
ByteBuf serRequest = MessageSerializer.serializeRequest(channel.alloc(), 282872L, request);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateEntry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateEntry.java
index 43b8e24..0567d13 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateEntry.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateEntry.java
@@ -43,7 +43,9 @@ public class KvStateEntry<K, N, V> {
private final ConcurrentMap<Thread, KvStateInfo<K, N, V>> serializerCache;
- public KvStateEntry(final InternalKvState<K, N, V> state) {
+ private final ClassLoader userClassLoader;
+
+ public KvStateEntry(final InternalKvState<K, N, V> state, ClassLoader userClassLoader) {
this.state = Preconditions.checkNotNull(state);
this.stateInfo =
new KvStateInfo<>(
@@ -51,6 +53,7 @@ public class KvStateEntry<K, N, V> {
state.getNamespaceSerializer(),
state.getValueSerializer());
this.serializerCache = new ConcurrentHashMap<>();
+ this.userClassLoader = userClassLoader;
this.areSerializersStateless = stateInfo.duplicate() == stateInfo;
}
@@ -58,6 +61,10 @@ public class KvStateEntry<K, N, V> {
return state;
}
+ public ClassLoader getUserClassLoader() {
+ return userClassLoader;
+ }
+
public KvStateInfo<K, N, V> getInfoForCurrentThread() {
return areSerializersStateless
? stateInfo
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java
index 1def92a..cf9c7bf 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/query/KvStateRegistry.java
@@ -87,11 +87,13 @@ public class KvStateRegistry {
JobVertexID jobVertexId,
KeyGroupRange keyGroupRange,
String registrationName,
- InternalKvState<?, ?, ?> kvState) {
+ InternalKvState<?, ?, ?> kvState,
+ ClassLoader userClassLoader) {
KvStateID kvStateId = new KvStateID();
- if (registeredKvStates.putIfAbsent(kvStateId, new KvStateEntry<>(kvState)) == null) {
+ if (registeredKvStates.putIfAbsent(kvStateId, new KvStateEntry<>(kvState, userClassLoader))
+ == null) {
final KvStateRegistryListener listener = getKvStateRegistryListener(jobId);
if (listener != null) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/query/TaskKvStateRegistry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/query/TaskKvStateRegistry.java
index f722402..cdd80158 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/query/TaskKvStateRegistry.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/query/TaskKvStateRegistry.java
@@ -60,10 +60,16 @@ public class TaskKvStateRegistry {
public void registerKvState(
KeyGroupRange keyGroupRange,
String registrationName,
- InternalKvState<?, ?, ?> kvState) {
+ InternalKvState<?, ?, ?> kvState,
+ ClassLoader userClassLoader) {
KvStateID kvStateId =
registry.registerKvState(
- jobId, jobVertexId, keyGroupRange, registrationName, kvState);
+ jobId,
+ jobVertexId,
+ keyGroupRange,
+ registrationName,
+ kvState,
+ userClassLoader);
registeredKvStates.add(new KvStateInfo(keyGroupRange, registrationName, kvStateId));
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
index 96dae88..210c789 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/AbstractKeyedStateBackend.java
@@ -283,7 +283,7 @@ public abstract class AbstractKeyedStateBackend<K>
throw new IllegalStateException("State backend has not been initialized for job.");
}
String name = stateDescriptor.getQueryableStateName();
- kvStateRegistry.registerKvState(keyGroupRange, name, kvState);
+ kvStateRegistry.registerKvState(keyGroupRange, name, kvState, userCodeClassLoader);
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
index 3f3ed74..f0ac718 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/query/KvStateRegistryTest.java
@@ -71,7 +71,12 @@ public class KvStateRegistryTest extends TestLogger {
final KvStateRegistry kvStateRegistry = new KvStateRegistry();
final KvStateID stateID =
kvStateRegistry.registerKvState(
- jobID, jobVertexId, keyGroupRange, registrationName, new DummyKvState());
+ jobID,
+ jobVertexId,
+ keyGroupRange,
+ registrationName,
+ new DummyKvState(),
+ getClass().getClassLoader());
final AtomicReference<Throwable> exceptionHolder = new AtomicReference<>();
@@ -162,7 +167,12 @@ public class KvStateRegistryTest extends TestLogger {
final String registrationName = "foobar";
final KvStateID kvStateID =
kvStateRegistry.registerKvState(
- jobId1, jobVertexId, keyGroupRange, registrationName, new DummyKvState());
+ jobId1,
+ jobVertexId,
+ keyGroupRange,
+ registrationName,
+ new DummyKvState(),
+ getClass().getClassLoader());
assertThat(registeredNotifications1.poll(), equalTo(jobId1));
assertThat(registeredNotifications2.isEmpty(), is(true));
@@ -176,7 +186,8 @@ public class KvStateRegistryTest extends TestLogger {
jobVertexId2,
keyGroupRange2,
registrationName2,
- new DummyKvState());
+ new DummyKvState(),
+ getClass().getClassLoader());
assertThat(registeredNotifications2.poll(), equalTo(jobId2));
assertThat(registeredNotifications1.isEmpty(), is(true));
@@ -222,7 +233,12 @@ public class KvStateRegistryTest extends TestLogger {
final String registrationName = "registrationName";
final KvStateID kvStateID =
kvStateRegistry.registerKvState(
- jobId, jobVertexId, keyGroupRange, registrationName, new DummyKvState());
+ jobId,
+ jobVertexId,
+ keyGroupRange,
+ registrationName,
+ new DummyKvState(),
+ getClass().getClassLoader());
assertThat(stateRegistrationNotifications.poll(), equalTo(jobId));
// another listener should not have received any notifications
[flink] 02/04: [FLINK-21138][tests] Register serializer instance in
AbstractQueryableStateTestBase.testCustomKryoSerializerHandling
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 5f6a804d11bcdf930a6a44ea48dd7e8f663954a3
Author: Till Rohrmann <tr...@apache.org>
AuthorDate: Mon Feb 8 19:28:55 2021 +0100
[FLINK-21138][tests] Register serializer instance in AbstractQueryableStateTestBase.testCustomKryoSerializerHandling
The test AbstractQueryableStateTestBase.testCustomKryoSerializerHandling only works if one registers an actual
serializer instance.
---
.../itcases/AbstractQueryableStateTestBase.java | 15 +++++++++++----
1 file changed, 11 insertions(+), 4 deletions(-)
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
index 2c0ca6c..7e8fa8f 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
@@ -85,6 +85,7 @@ import org.junit.Test;
import org.junit.rules.TemporaryFolder;
import java.io.IOException;
+import java.io.Serializable;
import java.net.URLClassLoader;
import java.time.Duration;
import java.util.ArrayList;
@@ -393,10 +394,7 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000L));
// Custom serializer is not needed, it's used just to check if serialization works.
- @SuppressWarnings("unchecked")
- Class<Serializer<?>> customSerializerClass =
- (Class<Serializer<?>>) userClassLoader.loadClass(customSerializerClassName);
- env.getConfig().addDefaultKryoSerializer(Byte.class, customSerializerClass);
+ env.getConfig().addDefaultKryoSerializer(Byte.class, createSerializer(userClassLoader));
// Here we *force* using Kryo, to check if custom serializers are handled correctly WRT
// classloading
@@ -432,6 +430,15 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
}
}
+ @SuppressWarnings("unchecked")
+ <T extends Serializer<Byte> & Serializable> T createSerializer(ClassLoader userClassLoader)
+ throws ClassNotFoundException, IllegalAccessException, InstantiationException {
+ Class<Serializer<?>> customSerializerClass =
+ (Class<Serializer<?>>) userClassLoader.loadClass("CustomKryo");
+
+ return (T) customSerializerClass.newInstance();
+ }
+
/**
* Tests that the correct exception is thrown if the query contains a wrong jobId or wrong
* queryable state name.
[flink] 03/04: [FLINK-21138] Fix classloader on QueryableStateClient
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 95281a7313d107e60a8103db3104a19cda5f9ece
Author: Maciek Próchniak <mp...@touk.pl>
AuthorDate: Tue Feb 9 16:59:56 2021 +0100
[FLINK-21138] Fix classloader on QueryableStateClient
---
.../client/QueryableStateClient.java | 8 +++++-
.../itcases/AbstractQueryableStateTestBase.java | 31 +++++++++++++---------
2 files changed, 26 insertions(+), 13 deletions(-)
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
index 8a575f0..395ddf0 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
@@ -47,6 +47,7 @@ import org.apache.flink.queryablestate.network.Client;
import org.apache.flink.queryablestate.network.messages.MessageSerializer;
import org.apache.flink.queryablestate.network.stats.DisabledKvStateRequestStats;
import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.LambdaUtil;
import org.apache.flink.util.NetUtils;
import org.apache.flink.util.Preconditions;
@@ -297,8 +298,13 @@ public class QueryableStateClient {
return FutureUtils.getFailedFuture(e);
}
+ ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
return getKvState(jobId, queryableStateName, key.hashCode(), serializedKeyAndNamespace)
- .thenApply(stateResponse -> createState(stateResponse, stateDescriptor));
+ .thenApply(
+ stateResponse ->
+ LambdaUtil.withContextClassLoader(
+ contextLoader,
+ () -> createState(stateResponse, stateDescriptor)));
}
private <T, S extends State> S createState(
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
index 7e8fa8f..faae032 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
@@ -394,7 +394,10 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, 1000L));
// Custom serializer is not needed, it's used just to check if serialization works.
- env.getConfig().addDefaultKryoSerializer(Byte.class, createSerializer(userClassLoader));
+ env.getConfig()
+ .addDefaultKryoSerializer(
+ Byte.class,
+ (Serializer<?> & Serializable) createSerializer(userClassLoader));
// Here we *force* using Kryo, to check if custom serializers are handled correctly WRT
// classloading
@@ -1480,6 +1483,7 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
if (!resultFuture.isDone()) {
CompletableFuture<S> expected =
client.getKvState(jobId, queryName, key, keyTypeInfo, stateDescriptor);
+ ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
expected.whenCompleteAsync(
(result, throwable) -> {
if (throwable != null) {
@@ -1490,17 +1494,20 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
instanceof UnknownKeyOrNamespaceException)) {
resultFuture.completeExceptionally(throwable.getCause());
} else if (deadline.hasTimeLeft()) {
- getKvStateIgnoringCertainExceptions(
- deadline,
- resultFuture,
- client,
- jobId,
- queryName,
- key,
- keyTypeInfo,
- stateDescriptor,
- failForUnknownKeyOrNamespace,
- executor);
+ LambdaUtil.withContextClassLoader(
+ contextLoader,
+ () ->
+ getKvStateIgnoringCertainExceptions(
+ deadline,
+ resultFuture,
+ client,
+ jobId,
+ queryName,
+ key,
+ keyTypeInfo,
+ stateDescriptor,
+ failForUnknownKeyOrNamespace,
+ executor));
}
} else {
resultFuture.complete(result);
[flink] 04/04: [FLINK-21138] Explicit Classloader in
QueryableStateClient
Posted by tr...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch release-1.11
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 30351d6a3d3f1a9949bf5f01e63293bf4a657424
Author: Maciek Próchniak <mp...@touk.pl>
AuthorDate: Wed Feb 10 09:11:37 2021 +0100
[FLINK-21138] Explicit Classloader in QueryableStateClient
---
.../client/QueryableStateClient.java | 22 ++++++++++--
.../itcases/AbstractQueryableStateTestBase.java | 39 ++++++++++------------
2 files changed, 38 insertions(+), 23 deletions(-)
diff --git a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
index 395ddf0..5920f24 100644
--- a/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
+++ b/flink-queryable-state/flink-queryable-state-client-java/src/main/java/org/apache/flink/queryablestate/client/QueryableStateClient.java
@@ -120,6 +120,9 @@ public class QueryableStateClient {
/** The execution configuration used to instantiate the different (de-)serializers. */
private ExecutionConfig executionConfig;
+ /** The user code classloader, used in loading serializers. */
+ private ClassLoader userClassLoader;
+
/**
* Create the Queryable State Client.
*
@@ -203,6 +206,18 @@ public class QueryableStateClient {
}
/**
+ * * Replaces the existing {@link ClassLoader} (possibly {@code null}), with the provided one.
+ *
+ * @param userClassLoader The new {@code userClassLoader}.
+ * @return The old classloader, or {@code null} if none was specified.
+ */
+ public ClassLoader setUserClassLoader(ClassLoader userClassLoader) {
+ ClassLoader prev = this.userClassLoader;
+ this.userClassLoader = userClassLoader;
+ return prev;
+ }
+
+ /**
* Returns a future holding the request result.
*
* @param jobId JobID of the job the queryable state belongs to.
@@ -298,12 +313,15 @@ public class QueryableStateClient {
return FutureUtils.getFailedFuture(e);
}
- ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
+ ClassLoader classLoaderToUse =
+ userClassLoader != null
+ ? userClassLoader
+ : Thread.currentThread().getContextClassLoader();
return getKvState(jobId, queryableStateName, key.hashCode(), serializedKeyAndNamespace)
.thenApply(
stateResponse ->
LambdaUtil.withContextClassLoader(
- contextLoader,
+ classLoaderToUse,
() -> createState(stateResponse, stateDescriptor)));
}
diff --git a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
index faae032..22b05bf 100644
--- a/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
+++ b/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/itcases/AbstractQueryableStateTestBase.java
@@ -72,7 +72,6 @@ import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
import org.apache.flink.testutils.ClassLoaderUtils;
import org.apache.flink.util.Collector;
import org.apache.flink.util.ExceptionUtils;
-import org.apache.flink.util.LambdaUtil;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.TestLogger;
@@ -425,11 +424,13 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
jobGraph.setClasspaths(Arrays.asList(userClassLoader.getURLs()));
clusterClient.submitJob(jobGraph).get();
- LambdaUtil.withContextClassLoader(
- userClassLoader,
- () ->
- executeValueQuery(
- deadline, client, jobId, stateName, valueState, numElements));
+
+ try {
+ client.setUserClassLoader(userClassLoader);
+ executeValueQuery(deadline, client, jobId, stateName, valueState, numElements);
+ } finally {
+ client.setUserClassLoader(null);
+ }
}
}
@@ -1483,7 +1484,6 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
if (!resultFuture.isDone()) {
CompletableFuture<S> expected =
client.getKvState(jobId, queryName, key, keyTypeInfo, stateDescriptor);
- ClassLoader contextLoader = Thread.currentThread().getContextClassLoader();
expected.whenCompleteAsync(
(result, throwable) -> {
if (throwable != null) {
@@ -1494,20 +1494,17 @@ public abstract class AbstractQueryableStateTestBase extends TestLogger {
instanceof UnknownKeyOrNamespaceException)) {
resultFuture.completeExceptionally(throwable.getCause());
} else if (deadline.hasTimeLeft()) {
- LambdaUtil.withContextClassLoader(
- contextLoader,
- () ->
- getKvStateIgnoringCertainExceptions(
- deadline,
- resultFuture,
- client,
- jobId,
- queryName,
- key,
- keyTypeInfo,
- stateDescriptor,
- failForUnknownKeyOrNamespace,
- executor));
+ getKvStateIgnoringCertainExceptions(
+ deadline,
+ resultFuture,
+ client,
+ jobId,
+ queryName,
+ key,
+ keyTypeInfo,
+ stateDescriptor,
+ failForUnknownKeyOrNamespace,
+ executor);
}
} else {
resultFuture.complete(result);