tzulitai commented on a change in pull request #8570: [FLINK-12688] [state]
Make serializer lazy initialization thread safe in StateDescriptor
URL: https://github.com/apache/flink/pull/8570#discussion_r289264163
##########
File path:
flink-core/src/test/java/org/apache/flink/api/common/state/StateDescriptorTest.java
##########
@@ -200,6 +204,31 @@ public void testEqualsSameNameAndTypeDifferentClass()
throws Exception {
assertNotEquals(descr1, descr2);
}
+ @Test
+ public void testSerializerLazyInitializeInParallel() throws Exception {
+ final String name = "testSerializerLazyInitializeInParallel";
+ // use PojoTypeInfo which will create a new serializer when
createSerializer is invoked.
+ final TestStateDescriptor<String> desc =
+ new TestStateDescriptor<>(name, new
PojoTypeInfo<>(String.class, new ArrayList<>()));
+ final int threadNumber = 20;
+ final ArrayList<Thread> threads = new ArrayList<>(threadNumber);
+ final ExecutionConfig executionConfig =
Mockito.mock(ExecutionConfig.class);
+ final ConcurrentHashMap<Integer, TypeSerializer<String>>
serializers = new ConcurrentHashMap<>();
+ for (int i = 0; i < threadNumber; i++) {
+ threads.add(new Thread(() -> {
+
desc.initializeSerializerUnlessSet(executionConfig);
+ TypeSerializer<String> serializer =
desc.getOriginalSerializer();
+
serializers.put(System.identityHashCode(serializer), serializer);
+ }));
+ }
+ threads.forEach(Thread::start);
+ for (Thread t : threads) {
+ t.join();
Review comment:
We should maybe also check for exceptions in these threads. You can use
Flink's `CheckedThread` for this purpose.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
