[jira] [Commented] (FLINK-16686) [State TTL] Make user class loader available in native RocksDB compaction thread

["Alexis Sarda-Espinosa (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-16686?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17761031#comment-17761031 ] 

Alexis Sarda-Espinosa commented on FLINK-16686:
-----------------------------------------------

cc [~yunta]

I wanted to check if this problem occurs with Avro, so I did a similar experiment as the one I did before with Kryo. I only considered {{GenericRecord}} serialization for now, but I can confirm this problem also occurs in that case.

I basically implemented a custom serializer, similar to what's described [in this example|https://medium.com/wbaa/evolve-your-data-model-in-flinks-state-using-avro-f26982afa399], and after running a dummy job for 2 hours I got this exception:

{noformat}
AvroEmbeddedRocksDBTest > avroSerialization() STANDARD_ERROR
    java.lang.NullPointerException
        at org.apache.avro.Schema.applyAliases(Schema.java:1872)
        at org.apache.avro.generic.GenericDatumReader.getResolver(GenericDatumReader.java:131)
        at org.apache.avro.generic.GenericDatumReader.read(GenericDatumReader.java:152)
        at com.avro.flink.GenericRecordTypeSerializer.deserialize(GenericRecordTypeSerializer.kt:56)
        at com.avro.flink.GenericRecordTypeSerializer.deserialize(GenericRecordTypeSerializer.kt:16)
        at org.apache.flink.api.common.typeutils.CompositeSerializer.deserialize(CompositeSerializer.java:156)
        at org.apache.flink.contrib.streaming.state.ttl.RocksDbTtlCompactFiltersManager$ListElementFilter.nextElementLastAccessTimestamp(RocksDbTtlCompactFiltersManager.java:205)
        at org.apache.flink.contrib.streaming.state.ttl.RocksDbTtlCompactFiltersManager$ListElementFilter.nextUnexpiredOffset(RocksDbTtlCompactFiltersManager.java:191)
{noformat}

Note that this stacktrace does _not_ mention a classloader. Some more details from the dump in case they are interesting:

{noformat}
Current thread (0x00007f7ae80b0800):  JavaThread "Thread-9949" [_thread_in_vm, id=9258, stack(0x00007f7a59a01000,0x00007f7a5a200000)]

Stack: [0x00007f7a59a01000,0x00007f7a5a200000],  sp=0x00007f7a5a1fcdc0,  free space=8175k
Native frames: (J=compiled Java code, A=aot compiled Java code, j=interpreted, Vv=VM code, C=native code)
V  [libjvm.so+0x7b8f75]
V  [libjvm.so+0x980aab]
C  [librocksdbjni-linux64.so+0x222ce1]  JavaListElementFilter::NextUnexpiredOffset(rocksdb::Slice const&, long, long) const+0x121
C  [librocksdbjni-linux64.so+0x648941]  rocksdb::flink::FlinkCompactionFilter::ListDecide(rocksdb::Slice const&, std::string*) const+0x81
C  [librocksdbjni-linux64.so+0x648de7]  rocksdb::flink::FlinkCompactionFilter::FilterV2(int, rocksdb::Slice const&, rocksdb::CompactionFilter::ValueType, rocksdb::Slice const&, std::string*, std::string*) const+0xc7
C  [librocksdbjni-linux64.so+0x3cce72]  rocksdb::MergeHelper::FilterMerge(rocksdb::Slice const&, rocksdb::Slice const&)+0x172
C  [librocksdbjni-linux64.so+0x3cdf06]  rocksdb::MergeHelper::MergeUntil(rocksdb::InternalIteratorBase<rocksdb::Slice>*, rocksdb::CompactionRangeDelAggregator*, unsigned long, bool, bool)+0xcc6
C  [librocksdbjni-linux64.so+0x2d2a81]  rocksdb::CompactionIterator::NextFromInput()+0x9a1
C  [librocksdbjni-linux64.so+0x2d4911]  rocksdb::CompactionIterator::SeekToFirst()+0x11
C  [librocksdbjni-linux64.so+0x2dd2be]  rocksdb::CompactionJob::ProcessKeyValueCompaction(rocksdb::CompactionJob::SubcompactionState*)+0x55e
C  [librocksdbjni-linux64.so+0x2de668]  rocksdb::CompactionJob::Run()+0x278
C  [librocksdbjni-linux64.so+0x33cec4]  rocksdb::DBImpl::BackgroundCompaction(bool*, rocksdb::JobContext*, rocksdb::LogBuffer*, rocksdb::DBImpl::PrepickedCompaction*, rocksdb::Env::Priority)+0x1084
C  [librocksdbjni-linux64.so+0x3407e7]  rocksdb::DBImpl::BackgroundCallCompaction(rocksdb::DBImpl::PrepickedCompaction*, rocksdb::Env::Priority)+0xd7
C  [librocksdbjni-linux64.so+0x340dda]  rocksdb::DBImpl::BGWorkCompaction(void*)+0x3a
C  [librocksdbjni-linux64.so+0x5e5744]  rocksdb::ThreadPoolImpl::Impl::BGThread(unsigned long)+0x254
C  [librocksdbjni-linux64.so+0x5e58ed]  rocksdb::ThreadPoolImpl::Impl::BGThreadWrapper(void*)+0x5d
{noformat}

> [State TTL] Make user class loader available in native RocksDB compaction thread
> --------------------------------------------------------------------------------
>
>                 Key: FLINK-16686
>                 URL: https://issues.apache.org/jira/browse/FLINK-16686
>             Project: Flink
>          Issue Type: Bug
>          Components: Runtime / State Backends
>    Affects Versions: 1.8.0, 1.11.3, 1.13.0, 1.12.3
>            Reporter: Andrey Zagrebin
>            Priority: Not a Priority
>              Labels: auto-deprioritized-major, auto-deprioritized-minor
>
> The issue is initially reported [here|https://stackoverflow.com/questions/60745711/flink-kryo-serializer-because-chill-serializer-couldnt-be-found].
> The problem is that the java code of Flink compaction filter is called from RocksDB native C++ code. It is called in the context of the native compaction thread. RocksDB has utilities to create java Thread context for the Flink java callback. Presumably, the Java thread context class loader is not set at all and if it is queried then it produces NullPointerException.
> The provided report enabled a list state with TTL. The compaction filter has to deserialise elements to check expiration. The deserialiser relies on Kryo which queries the thread context class loader which is expected to be the user class loader of the task but turns out to be null.
> We should investigate how to pass the user class loader to the compaction thread of the list state with TTL.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)