[ https://issues.apache.org/jira/browse/FLINK-9693?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16532514#comment-16532514 ]
ASF GitHub Bot commented on FLINK-9693: --------------------------------------- Github user zentol commented on a diff in the pull request: https://github.com/apache/flink/pull/6251#discussion_r200062854 --- Diff: flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionTest.java --- @@ -385,6 +381,76 @@ public void testTerminationFutureIsCompletedAfterSlotRelease() throws Exception restartFuture.get(); } + /** + * Tests that the task restore state is nulled after the {@link Execution} has been + * deployed. See FLINK-9693. + */ + @Test + public void testTaskRestoreStateIsNulledAfterDeployment() throws Exception { + final JobVertex jobVertex = createNoOpJobVertex(); + final JobVertexID jobVertexId = jobVertex.getID(); + + final SingleSlotTestingSlotOwner slotOwner = new SingleSlotTestingSlotOwner(); + final ProgrammedSlotProvider slotProvider = createProgrammedSlotProvider( + 1, + Collections.singleton(jobVertexId), + slotOwner); + + ExecutionGraph executionGraph = ExecutionGraphTestUtils.createSimpleTestGraph( + new JobID(), + slotProvider, + new NoRestartStrategy(), + jobVertex); + + ExecutionJobVertex executionJobVertex = executionGraph.getJobVertex(jobVertexId); + + ExecutionVertex executionVertex = executionJobVertex.getTaskVertices()[0]; + + final Execution execution = executionVertex.getCurrentExecutionAttempt(); + + final JobManagerTaskRestore taskRestoreState = new JobManagerTaskRestore(1L, new TaskStateSnapshot()); + execution.setInitialState(taskRestoreState); + + assertThat(execution.getTaskRestore(), is(notNullValue())); + + // schedule the execution vertex and wait for its deployment + executionVertex.scheduleForExecution(slotProvider, false, LocationPreferenceConstraint.ANY).get(); + + assertThat(execution.getTaskRestore(), is(nullValue())); + } + + @Nonnull + private JobVertex createNoOpJobVertex() { + final JobVertex jobVertex = new JobVertex("Test vertex", new JobVertexID()); + jobVertex.setInvokableClass(NoOpInvokable.class); + + return jobVertex; + } + + @Nonnull + private ProgrammedSlotProvider createProgrammedSlotProvider( + int parallelism, + Collection<JobVertexID> jobVertexIds, + SlotOwner slotOwner) { + final ProgrammedSlotProvider slotProvider = new ProgrammedSlotProvider(parallelism); + + for (JobVertexID jobVertexId : jobVertexIds) { + for (int i = 0; i < parallelism; i++) { + final SimpleSlot slot = new SimpleSlot( + slotOwner, + new LocalTaskManagerLocation(), + 0, + new SimpleAckingTaskManagerGateway(), + null, + null ); --- End diff -- whitepace after `null` > Possible memory leak in jobmanager retaining archived checkpoints > ----------------------------------------------------------------- > > Key: FLINK-9693 > URL: https://issues.apache.org/jira/browse/FLINK-9693 > Project: Flink > Issue Type: Bug > Components: JobManager, State Backends, Checkpointing > Affects Versions: 1.5.0, 1.6.0 > Environment: !image.png!!image (1).png! > Reporter: Steven Zhen Wu > Assignee: Till Rohrmann > Priority: Major > Labels: pull-request-available > Attachments: 41K_ExecutionVertex_objs_retained_9GB.png, > ExecutionVertexZoomIn.png > > > First, some context about the job > * Flink 1.4.1 > * stand-alone deployment mode > * embarrassingly parallel: all operators are chained together > * parallelism is over 1,000 > * stateless except for Kafka source operators. checkpoint size is 8.4 MB. > * set "state.backend.fs.memory-threshold" so that only jobmanager writes to > S3 to checkpoint > * internal checkpoint with 10 checkpoints retained in history > > Summary of the observations > * 41,567 ExecutionVertex objects retained 9+ GB of memory > * Expanded in one ExecutionVertex. it seems to storing the kafka offsets for > source operator -- This message was sent by Atlassian JIRA (v7.6.3#76005)