This is an automated email from the ASF dual-hosted git repository.
zakelly pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 4a3e60da78b [FLINK-16686][State] Set classloader for compaction
filters (#27073)
4a3e60da78b is described below
commit 4a3e60da78b5ba389ab41c71b79d60ddc13fd24e
Author: Zakelly <[email protected]>
AuthorDate: Fri Oct 3 13:01:30 2025 +0800
[FLINK-16686][State] Set classloader for compaction filters (#27073)
---
.../state/ttl/TtlListStateWithKryoTestContext.java | 93 ++++++++++++++++++++++
.../flink/runtime/state/ttl/TtlStateTestBase.java | 1 +
.../state/ttl/RocksDbTtlCompactFiltersManager.java | 64 ++++++++++++---
3 files changed, 149 insertions(+), 9 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlListStateWithKryoTestContext.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlListStateWithKryoTestContext.java
new file mode 100644
index 00000000000..0796a996b42
--- /dev/null
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlListStateWithKryoTestContext.java
@@ -0,0 +1,93 @@
+/*
+ * 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.runtime.state.ttl;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.serialization.SerializerConfig;
+import org.apache.flink.api.common.serialization.SerializerConfigImpl;
+import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.PipelineOptions;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Test suite for {@link TtlListState} with elements of serialized by kryo. */
+public class TtlListStateWithKryoTestContext
+ extends
TtlListStateTestContextBase<TtlListStateWithKryoTestContext.NotPojoElement> {
+ TtlListStateWithKryoTestContext() {
+ super(new KryoSerializer<>(NotPojoElement.class,
getForceKryoSerializerConfig()));
+ }
+
+ private static SerializerConfig getForceKryoSerializerConfig() {
+ Configuration config = new Configuration();
+ config.set(PipelineOptions.FORCE_KRYO, true);
+ return new SerializerConfigImpl(config, new ExecutionConfig(config));
+ }
+
+ @Override
+ NotPojoElement generateRandomElement(int i) {
+ return new NotPojoElement(RANDOM.nextInt(100));
+ }
+
+ @Override
+ void initTestValues() {
+ emptyValue = Collections.emptyList();
+
+ updateEmpty =
+ Arrays.asList(new NotPojoElement(5), new NotPojoElement(7),
new NotPojoElement(10));
+ updateUnexpired =
+ Arrays.asList(new NotPojoElement(8), new NotPojoElement(9),
new NotPojoElement(11));
+ updateExpired = Arrays.asList(new NotPojoElement(1), new
NotPojoElement(4));
+
+ getUpdateEmpty = updateEmpty;
+ getUnexpired = updateUnexpired;
+ getUpdateExpired = updateExpired;
+ }
+
+ public static class NotPojoElement {
+ public int value;
+
+ public NotPojoElement(int value) {
+ this.value = value;
+ }
+
+ @Override
+ public String toString() {
+ return "NotPojoElement{" + "value=" + value + '}';
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ NotPojoElement that = (NotPojoElement) obj;
+ return value == that.value;
+ }
+
+ @Override
+ public int hashCode() {
+ return Integer.hashCode(value);
+ }
+ }
+}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
index bd658dbea44..d538738c28b 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/state/ttl/TtlStateTestBase.java
@@ -87,6 +87,7 @@ public abstract class TtlStateTestBase {
new TtlValueStateTestContext(),
new TtlFixedLenElemListStateTestContext(),
new TtlNonFixedLenElemListStateTestContext(),
+ new TtlListStateWithKryoTestContext(),
new TtlMapStateAllEntriesTestContext(),
new TtlMapStatePerElementTestContext(),
new TtlMapStatePerNullElementTestContext(),
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
index 168da2f2af1..e475c5cf67b 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/contrib/streaming/state/ttl/RocksDbTtlCompactFiltersManager.java
@@ -47,6 +47,7 @@ import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.LinkedHashMap;
+import java.util.function.Supplier;
/** RocksDB compaction filter utils for state with TTL. */
public class RocksDbTtlCompactFiltersManager {
@@ -160,15 +161,27 @@ public class RocksDbTtlCompactFiltersManager {
private static class ListElementFilterFactory<T>
implements FlinkCompactionFilter.ListElementFilterFactory {
- private final TypeSerializer<T> serializer;
+ // {@See #createListElementFilter}.
+ private final ThreadLocalSerializerProvider<T> threadLocalSerializer;
private ListElementFilterFactory(TypeSerializer<T> serializer) {
- this.serializer = serializer;
+ ClassLoader contextClassLoader = null;
+ try {
+ contextClassLoader =
Thread.currentThread().getContextClassLoader();
+ } catch (Throwable e) {
+ LOG.info("Cannot get context classloader for list state's
compaction filter.", e);
+ }
+ threadLocalSerializer =
+ new ThreadLocalSerializerProvider<>(serializer,
contextClassLoader);
}
@Override
public FlinkCompactionFilter.ListElementFilter
createListElementFilter() {
- return new ListElementFilter<>(serializer);
+ // This method will be invoked by native code multiple times when
creating compaction
+ // filter. And the created filter will be shared by multiple
background threads.
+ // Make sure the serializer is thread-local and has classloader
set for each thread
+ // correctly and individually.
+ return new ListElementFilter<>(threadLocalSerializer);
}
}
@@ -186,11 +199,11 @@ public class RocksDbTtlCompactFiltersManager {
}
private static class ListElementFilter<T> implements
FlinkCompactionFilter.ListElementFilter {
- private final TypeSerializer<T> serializer;
- private DataInputDeserializer input;
+ private final ThreadLocalSerializerProvider<T> threadLocalSerializer;
+ private final DataInputDeserializer input;
- private ListElementFilter(TypeSerializer<T> serializer) {
- this.serializer = serializer;
+ private ListElementFilter(ThreadLocalSerializerProvider<T> serializer)
{
+ this.threadLocalSerializer = serializer;
this.input = new DataInputDeserializer();
}
@@ -198,9 +211,10 @@ public class RocksDbTtlCompactFiltersManager {
public int nextUnexpiredOffset(byte[] bytes, long ttl, long
currentTimestamp) {
input.setBuffer(bytes);
int lastElementOffset = 0;
+ TypeSerializer<T> serializer = threadLocalSerializer.get();
while (input.available() > 0) {
try {
- long timestamp = nextElementLastAccessTimestamp();
+ long timestamp =
nextElementLastAccessTimestamp(serializer);
if (!TtlUtils.expired(timestamp, ttl, currentTimestamp)) {
break;
}
@@ -213,7 +227,8 @@ public class RocksDbTtlCompactFiltersManager {
return lastElementOffset;
}
- private long nextElementLastAccessTimestamp() throws IOException {
+ private long nextElementLastAccessTimestamp(TypeSerializer<T>
serializer)
+ throws IOException {
TtlValue<?> ttlValue = (TtlValue<?>) serializer.deserialize(input);
if (input.available() > 0) {
input.skipBytesToRead(1);
@@ -222,6 +237,37 @@ public class RocksDbTtlCompactFiltersManager {
}
}
+ private static class ThreadLocalSerializerProvider<T> implements
Supplier<TypeSerializer<T>> {
+ // Multiple background threads may share the same filter instance, so
we need to make sure
+ // the serializer is thread-local, and every thread has its own
instance with classloader.
+ private final ThreadLocal<TypeSerializer<T>> threadLocalSerializer;
+
+ public ThreadLocalSerializerProvider(
+ TypeSerializer<T> serializer, ClassLoader classLoader) {
+ this.threadLocalSerializer =
+ ThreadLocal.withInitial(
+ () -> {
+ setClassloaderIfNeeded(classLoader);
+ return serializer.duplicate();
+ });
+ }
+
+ private void setClassloaderIfNeeded(ClassLoader classLoader) {
+ // The classloader that should be set to the current thread when
deserializing.
+ // The reason why we should set classloader is that the serializer
may be Kryo
+ // serializer which needs user classloader to load user classes.
+ // See FLINK-16686 for more details.
+ if (classLoader != null) {
+ Thread.currentThread().setContextClassLoader(classLoader);
+ }
+ }
+
+ @Override
+ public TypeSerializer<T> get() {
+ return threadLocalSerializer.get();
+ }
+ }
+
public void disposeAndClearRegisteredCompactionFactories() {
for (FlinkCompactionFilterFactory factory :
compactionFilterFactories.values()) {
IOUtils.closeQuietly(factory);
