[FLINK-8005] [runtime] Set user code class loader before snapshot by GJL · Pull Request #4980 · apache/flink

GJL · 2017-11-08T10:58:22Z

What is the purpose of the change

During checkpointing, user code may dynamically load classes from the user code
jar. This is a problem if the thread invoking the snapshot callbacks does not
have the user code class loader set as its context class loader. This commit
makes sure that the correct class loader is set.

Brief change log

Set user code class loader in ThreadFactory of Task#asyncCallDispatcher
Clean up TaskAsyncCallTest

Verifying this change

This change added tests and can be verified as follows:

Added unit tests to verify that context class loader is set
Started job with FlinkKafkaProducer011 and verified that snapshotting works

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (yes / no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes / no)
The serializers: (yes / no / don't know)
The runtime per-record code paths (performance sensitive): (yes / no / don't know)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: (yes / no / don't know)
The S3 file system connector: (yes / no / don't know)

Documentation

Does this pull request introduce a new feature? (yes / no)
If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)

During checkpointing, user code may dynamically load classes from the user code jar. This is a problem if the thread invoking the snapshot callbacks does not have the user code class loader set as its context class loader. This commit makes sure that the correct class loader is set.

GJL · 2017-11-08T11:04:02Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

+		for (int i = 1; i <= numCalls; i++) {
+			task.triggerCheckpointBarrier(i, 156865867234L, CheckpointOptions.forCheckpoint());
 		}
-		catch (Exception e) {


Semantics of existing tests did not change: I removed the try-catch and simplified the assertion:

if (currentState != ExecutionState.RUNNING && currentState != ExecutionState.FINISHED) { fail("Task should be RUNNING or FINISHED, but is " + currentState); }

to

assertThat(currentState, isOneOf(ExecutionState.RUNNING, ExecutionState.FINISHED));

Yep, the diff on GitHub is a bit hard to read but I figured it out. 😅

aljoscha · 2017-11-08T11:04:06Z

These changes look good! 👍

I'll wait for travis and then merge.

GJL · 2017-11-08T15:32:42Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskStopTest.java

-	}
-
-	@Test(timeout = 20000)
-	public void testStopExecution() throws Exception {


This only tested that stop is invoked. Should be covered by testSetsUserCodeClassLoader now.

GJL · 2017-11-08T15:33:05Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskStopTest.java

-	}
-
-	@Test(expected = RuntimeException.class)
-	public void testStopExecutionFail() throws Exception {


This is now covered by testThrowExceptionIfStopInvokedWithNotStoppableTask

GJL · 2017-11-08T15:34:28Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

-				fail("Task should be RUNNING or FINISHED, but is " + currentState);
-			}
-
-			task.cancelExecution();


I moved this to the AutoCloseable TaskCleaner to avoid duplication.

aljoscha · 2017-11-09T10:47:08Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

 		}
+
+		// assert after task is canceled and executing thread is stopped to avoid race conditions
+		assertThat(classLoaders, hasSize(greaterThanOrEqualTo(3)));


Are we guaranteed that all three calls have been made at this point or could this be flaky due to race conditions?

As @aljoscha suggested, I think that there is no guarantee that the 3 calls have finished by the time we check, right?

I believe you are right. I introduced another latch to counter this.

kl0u

Nice work @GJL ! I have some minor comments, and a question about a test. As soon as these are addressed, I think that this is good to merge.

kl0u · 2017-11-09T12:23:21Z

flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/DispatcherThreadFactory.java

+	 * @param classLoader The {@link ClassLoader} to be set as context class loader.
+	 */
+	public DispatcherThreadFactory(
+		ThreadGroup group,


This is a code style preference rather than an issue, but I would suggest to indent the arguments by a tab to separate them from the body of the method.

It is Indented now.

kl0u · 2017-11-09T12:24:07Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java


 		@Override
 		public void abortCheckpointOnBarrier(long checkpointId, Throwable cause) {
-			throw new UnsupportedOperationException("Should not be called");


Why was this removed?

Not sure anymore but I decided to add it again.

kl0u · 2017-11-09T12:24:16Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java


 		@Override
 		public void triggerCheckpointOnBarrier(CheckpointMetaData checkpointMetaData, CheckpointOptions checkpointOptions, CheckpointMetrics checkpointMetrics) throws Exception {
-			throw new UnsupportedOperationException("Should not be called");


Why was this removed?

Not sure anymore but I decided to add it again.

kl0u · 2017-11-09T12:35:18Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

 		}
+
+		// assert after task is canceled and executing thread is stopped to avoid race conditions
+		assertThat(classLoaders, hasSize(greaterThanOrEqualTo(3)));


As @aljoscha suggested, I think that there is no guarantee that the 3 calls have finished by the time we check, right?

Throw UnsupportedOperationException when CheckpointsInOrderInvokable#triggerCheckpointOnBarrier() and CheckpointsInOrderInvokable#abortCheckpointOnBarrier() are called.

GJL · 2017-11-09T13:08:38Z

I addressed the comments. Let's wait for Travis and let me know if something else needs to be changed.

@aljoscha @kl0u

aljoscha · 2017-11-09T13:20:59Z

I think waiting on the stop latch might not be enough (in 100 % of cases) because the other two calls are also asynchronous.

GJL · 2017-11-09T13:25:20Z

As it is now, it should be enough as there is only one thread dispatching the calls:

executor = Executors.newSingleThreadExecutor(
						new DispatcherThreadFactory(TASK_THREADS_GROUP, "Async calls on " + taskNameWithSubtask));
				this.asyncCallDispatcher = executor;

The tasks cannot overtake each other. I could make the test stricter and wait additionally on triggerLatch in case somebody decides to have multiple threads.

@aljoscha

aljoscha · 2017-11-09T13:56:36Z

Yes, but I think this is making an assumption about the internal implementation. If someone changes that the test could break/not test the right thing anymore.

kl0u · 2017-11-09T14:52:57Z

flink-runtime/src/test/java/org/apache/flink/runtime/taskmanager/TaskAsyncCallTest.java

-			triggerLatch.trigger();
 			if (error != null) {
+				// exit method prematurely due to error but make sure that the tests can finish
+				triggerLatch.trigger();


for all latches, it should also have:if (!latch.isTriggered()) { latch.await() }

Why is that? I think at this point the latch might not get triggered at all (except here).

Sorry, I was just looking on the IDE and missed the lines. This line should be before every time you call await on the latch.

I think it doesn't matter because the latch already checks for the flag:

public void await() throws InterruptedException { synchronized (lock) { while (!triggered) { lock.wait(); } } }

public void trigger() { synchronized (lock) { triggered = true; lock.notifyAll(); } }

yes, a latch that was already triggered will simply return immediately, no need for an additional check

aljoscha · 2017-11-09T15:44:53Z

thanks, I think this is excellent now. 👌

I'll merge as soon as travis is green.

kl0u · 2017-11-09T15:53:15Z

I agree! +1 to merge as soon as Travis gives us the green light.

aljoscha · 2017-11-10T08:27:50Z

Thanks again for this fix! 👍

Could you please close if GitHub doesn't auto-close?

GJL commented Nov 8, 2017

View reviewed changes

GJL closed this Nov 8, 2017

[FLINK-8005] [runtime] Move tests in TaskStopTest to TaskAsyncCallTest

e5c5a42

GJL reopened this Nov 8, 2017

GJL commented Nov 8, 2017

View reviewed changes

aljoscha reviewed Nov 9, 2017

View reviewed changes

kl0u reviewed Nov 9, 2017

View reviewed changes

GJL added 3 commits November 9, 2017 13:59

[FLINK-8005] [runtime] Fix possible race condition in TaskAsyncCallTest

b7da56b

[FLINK-8005] [runtime] Update TaskAsyncCallTest

53efe5d

Throw UnsupportedOperationException when CheckpointsInOrderInvokable#triggerCheckpointOnBarrier() and CheckpointsInOrderInvokable#abortCheckpointOnBarrier() are called.

[FLINK-8005] [runtime] Re-format DispatcherThreadFactory constructor

d6b33d2

[FLINK-8005] [runtime] Fix possible race condition in TaskAsyncCallTest

f53f7c3

kl0u reviewed Nov 9, 2017

View reviewed changes

GJL closed this Nov 10, 2017

rmetzger added component=Connectors/Kafka component=Runtime/StateBackends labels Mar 18, 2019

Conversation

GJL commented Nov 8, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljoscha commented Nov 8, 2017

Uh oh!

GJL Nov 8, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Nov 8, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kl0u left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Nov 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Nov 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL commented Nov 9, 2017

Uh oh!

aljoscha commented Nov 9, 2017

Uh oh!

GJL commented Nov 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aljoscha commented Nov 9, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Nov 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

GJL Nov 9, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

aljoscha commented Nov 9, 2017

Uh oh!

kl0u commented Nov 9, 2017

Uh oh!

aljoscha commented Nov 10, 2017

GJL commented Nov 8, 2017 •

edited

Loading

GJL Nov 8, 2017 •

edited

Loading

GJL Nov 8, 2017 •

edited

Loading

GJL Nov 9, 2017 •

edited

Loading

GJL Nov 9, 2017 •

edited

Loading

GJL commented Nov 9, 2017 •

edited

Loading

GJL Nov 9, 2017 •

edited

Loading

GJL Nov 9, 2017 •

edited

Loading