[jira] [Created] (FLINK-11250) fix thread lack when StreamTask switched from DEPLOYING to CANCELING

[Shang Yuanchun (Jira)]

lamber-ken created FLINK-11250:
----------------------------------

             Summary: fix thread lack when StreamTask switched from DEPLOYING to CANCELING
                 Key: FLINK-11250
                 URL: https://issues.apache.org/jira/browse/FLINK-11250
             Project: Flink
          Issue Type: Bug
          Components: Local Runtime, Streaming
    Affects Versions: 1.7.1, 1.6.3, 1.5.6
            Reporter: lamber-ken
            Assignee: lamber-ken
             Fix For: 1.7.2


begin flink-1.5.x version, streamRecordWriters was created in StreamTask's constructor, which start OutputFlusher daemon thread. so when task switched from DEPLOYING to CANCELING state, the daemon thread will be lacked.

 

*reproducible example*
{code:java}
public static void main(String[] args) throws Exception {

    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    env.enableCheckpointing(5000);
    env
            .addSource(new SourceFunction<String>() {
                @Override
                public void run(SourceContext<String> ctx) throws Exception {

                    for (int i = 0; i < 100000000; i++) {
                        Thread.sleep(100);
                        ctx.collect("data " + i);
                    }
                }

                @Override
                public void cancel() {

                }
            })
            .addSink(new RichSinkFunction<String>() {

                @Override
                public void open(Configuration parameters) throws Exception {
                    System.out.println(1 / 0);
                }

                @Override
                public void invoke(String value, Context context) throws Exception {

                }

            }).setParallelism(2);


    env.execute();

}{code}
*some useful log*
{code:java}
2019-01-02 03:03:47.525 [thread==> jobmanager-future-thread-2] executiongraph.Execution#transitionState:1316 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from CREATED to SCHEDULED.
2019-01-02 03:03:47.526 [thread==> flink-akka.actor.default-dispatcher-5] slotpool.SlotPool#allocateSlot:326 Received slot request [SlotRequestId{12bfcf1674f5b96567a076086dbbfd1b}] for task: Attempt #1 (Source: Custom Source (1/1)) @ (unassigned) - [SCHEDULED]
2019-01-02 03:03:47.527 [thread==> flink-akka.actor.default-dispatcher-5] slotpool.SlotSharingManager#createRootSlot:151 Create multi task slot [SlotRequestId{494e47eb8318e2c09999a1db91dda6b8}] in slot [SlotRequestId{6d7f0173c1d48e5559f6a14b080ee817}].
2019-01-02 03:03:47.527 [thread==> flink-akka.actor.default-dispatcher-5] slotpool.SlotSharingManager$MultiTaskSlot#allocateSingleTaskSlot:426 Create single task slot [SlotRequestId{12bfcf1674f5b96567a076086dbbfd1b}] in multi task slot [SlotRequestId{494e47eb8318e2c09999a1db91dda6b8}] for group bc764cd8ddf7a0cff126f51c16239658.
2019-01-02 03:03:47.528 [thread==> flink-akka.actor.default-dispatcher-2] slotpool.SlotSharingManager$MultiTaskSlot#allocateSingleTaskSlot:426 Create single task slot [SlotRequestId{8a877431375df8aeadb2fd845cae15fc}] in multi task slot [SlotRequestId{494e47eb8318e2c09999a1db91dda6b8}] for group 0a448493b4782967b150582570326227.
2019-01-02 03:03:47.528 [thread==> flink-akka.actor.default-dispatcher-2] slotpool.SlotSharingManager#createRootSlot:151 Create multi task slot [SlotRequestId{56a36d3902ee1a7d0e2e84f50039c1ca}] in slot [SlotRequestId{dbf5c9fa39f1e5a0b34a4a8c10699ee5}].
2019-01-02 03:03:47.528 [thread==> flink-akka.actor.default-dispatcher-2] slotpool.SlotSharingManager$MultiTaskSlot#allocateSingleTaskSlot:426 Create single task slot [SlotRequestId{5929c12b52dccee682f86afbe1cff5cf}] in multi task slot [SlotRequestId{56a36d3902ee1a7d0e2e84f50039c1ca}] for group 0a448493b4782967b150582570326227.
2019-01-02 03:03:47.529 [thread==> flink-akka.actor.default-dispatcher-5] executiongraph.Execution#transitionState:1316 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from SCHEDULED to DEPLOYING.
2019-01-02 03:03:47.529 [thread==> flink-akka.actor.default-dispatcher-5] executiongraph.Execution#deploy:576 Deploying Source: Custom Source (1/1) (attempt #1) to localhost
2019-01-02 03:03:47.530 [thread==> flink-akka.actor.default-dispatcher-2] state.TaskExecutorLocalStateStoresManager#localStateStoreForSubtask:162 Registered new local state store with configuration LocalRecoveryConfig{localRecoveryMode=false, localStateDirectories=LocalRecoveryDirectoryProvider{rootDirectories=[/tmp/localState/aid_AllocationID{7b5faad9073d7fac6759e40981197b8d}], jobID=06e76f6e31728025b22fdda9fadd6f01, jobVertexID=bc764cd8ddf7a0cff126f51c16239658, subtaskIndex=0}} for 06e76f6e31728025b22fdda9fadd6f01 - bc764cd8ddf7a0cff126f51c16239658 - 0 under allocation id AllocationID{7b5faad9073d7fac6759e40981197b8d}.
2019-01-02 03:03:47.530 [thread==> flink-akka.actor.default-dispatcher-2] partition.ResultPartition#<init>:172 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef): Initialized ResultPartition 85c7415bae559d6198b8bb69d4c6e49f@74a4ed4bb2f80aa2b98e11bd09ea64ef [PIPELINED_BOUNDED, 2 subpartitions, 2 pending references]
2019-01-02 03:03:47.530 [thread==> flink-akka.actor.default-dispatcher-2] taskexecutor.TaskExecutor#submitTask:541 Received task Source: Custom Source (1/1).
2019-01-02 03:03:47.532 [thread==> Source: Custom Source (1/1)] taskmanager.Task#transitionState:992 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from CREATED to DEPLOYING.
2019-01-02 03:03:47.532 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:655 Creating FileSystem stream leak safety net for task Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING]
2019-01-02 03:03:47.532 [thread==> flink-akka.actor.default-dispatcher-2] state.TaskExecutorLocalStateStoresManager#localStateStoreForSubtask:166 Found existing local state store for 06e76f6e31728025b22fdda9fadd6f01 - 0a448493b4782967b150582570326227 - 0 under allocation id AllocationID{7b5faad9073d7fac6759e40981197b8d}: org.apache.flink.runtime.state.NoOpTaskLocalStateStoreImpl@176e8a65
2019-01-02 03:03:47.535 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:662 Loading JAR files for task Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.536 [thread==> Source: Custom Source (1/1)] taskmanager.Task#createUserCodeClassloader:947 Getting user code class loader for task 74a4ed4bb2f80aa2b98e11bd09ea64ef at library cache manager took 0 milliseconds
2019-01-02 03:03:47.537 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:688 Registering task at network: Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.537 [thread==> Source: Custom Source (1/1)] buffer.LocalBufferPool#<init>:125 Using a local buffer pool with 2-12 buffers
2019-01-02 03:03:47.537 [thread==> Source: Custom Source (1/1)] partition.ResultPartitionManager#registerResultPartition:62 Registered ResultPartition 85c7415bae559d6198b8bb69d4c6e49f@74a4ed4bb2f80aa2b98e11bd09ea64ef [PIPELINED_BOUNDED, 2 subpartitions, 2 pending references].
2019-01-02 03:03:47.537 [thread==> Source: Custom Source (1/1)] network.TaskEventDispatcher#registerPartition:59 registering 85c7415bae559d6198b8bb69d4c6e49f@74a4ed4bb2f80aa2b98e11bd09ea64ef
2019-01-02 03:03:47.537 [thread==> flink-akka.actor.default-dispatcher-2] state.TaskExecutorLocalStateStoresManager#localStateStoreForSubtask:166 Found existing local state store for 06e76f6e31728025b22fdda9fadd6f01 - 0a448493b4782967b150582570326227 - 1 under allocation id AllocationID{d8a1fd06e3e52101b003a04355f64be1}: org.apache.flink.runtime.state.NoOpTaskLocalStateStoreImpl@57bf2d8e
2019-01-02 03:03:47.537 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:714 =======> next, kick off the background copying of files for the distributed cache: Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.538 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:729 =======> isCanceledOrFailed: Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.540 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:737 =======> call the user code initialization methods: Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.543 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:765 =======> now load and instantiate the task's invokable code: Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [DEPLOYING].
2019-01-02 03:03:47.543 [thread==> flink-akka.actor.default-dispatcher-5] executiongraph.ExecutionGraph#transitionState:1356 Job Flink Streaming Job (06e76f6e31728025b22fdda9fadd6f01) switched from state RUNNING to FAILING.
2019-01-02 03:03:47.544 [thread==> Source: Custom Source (1/1)] tasks.StreamTask#createStreamRecordWriter:1214 Using partitioner REBALANCE for output 0 of task Source: Custom Source
2019-01-02 03:03:47.544 [thread==> Source: Custom Source (1/1)] io.StreamRecordWriter$OutputFlusher#<init>:174 StreamRecordWriter start : outputflusher1546369427544, timeout: 100
2019-01-02 03:03:47.544 [thread==> flink-akka.actor.default-dispatcher-5] executiongraph.Execution#transitionState:1316 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from DEPLOYING to CANCELING.
2019-01-02 03:03:47.544 [thread==> flink-akka.actor.default-dispatcher-2] taskmanager.Task#cancelExecution:1055 Attempting to cancel task Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef).
2019-01-02 03:03:47.544 [thread==> flink-akka.actor.default-dispatcher-2] taskmanager.Task#cancelOrFailAndCancelInvokable:1078 cancelOrFailAndCancelInvokable_current_execution_state==> task: Source: Custom Source (1/1), state: DEPLOYING, targetState CANCELING
2019-01-02 03:03:47.545 [thread==> flink-akka.actor.default-dispatcher-2] taskmanager.Task#transitionState:992 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from DEPLOYING to CANCELING.
2019-01-02 03:03:47.545 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#releaseSlot:745 Releasing slot [SlotRequestId{5929c12b52dccee682f86afbe1cff5cf}] because: Slot is being returned to the SlotPool.
2019-01-02 03:03:47.545 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#releaseSlot:745 Releasing slot [SlotRequestId{dbf5c9fa39f1e5a0b34a4a8c10699ee5}] because: Release multi task slot because all children have been released.
2019-01-02 03:03:47.545 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#tryFulfillSlotRequestOrMakeAvailable:842 Adding returned slot [AllocationID{d8a1fd06e3e52101b003a04355f64be1}] to available slots
2019-01-02 03:03:47.546 [thread==> flink-akka.actor.default-dispatcher-2] taskexecutor.TaskExecutor#failPartition:656 Discarding the results produced by task execution 8aeff98e78ed3d348af9d4d5679f2b26.
2019-01-02 03:03:47.546 [thread==> flink-akka.actor.default-dispatcher-2] partition.ResultPartitionManager#releasePartitionsProducedBy:105 Released all partitions produced by 8aeff98e78ed3d348af9d4d5679f2b26.
2019-01-02 03:03:47.547 [thread==> Source: Custom Source (1/1)] taskmanager.Task#transitionState:992 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from CANCELING to CANCELED.
2019-01-02 03:03:47.547 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:892 Freeing task resources for Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef).
2019-01-02 03:03:47.547 [thread==> Source: Custom Source (1/1)] network.NetworkEnvironment#unregisterTask:263 Unregister task Source: Custom Source (1/1) from network environment (state: CANCELED).
2019-01-02 03:03:47.547 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:47.547 [thread==> Source: Custom Source (1/1)] partition.ResultPartition#release:313 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef): Releasing ResultPartition 85c7415bae559d6198b8bb69d4c6e49f@74a4ed4bb2f80aa2b98e11bd09ea64ef [PIPELINED_BOUNDED, 2 subpartitions, 2 pending references].
2019-01-02 03:03:47.548 [thread==> Source: Custom Source (1/1)] partition.PipelinedSubpartition#release:137 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef): Released PipelinedSubpartition#0 [number of buffers: 0 (0 bytes), number of buffers in backlog: 0, finished? false, read view? false].
2019-01-02 03:03:47.548 [thread==> Source: Custom Source (1/1)] partition.PipelinedSubpartition#release:137 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef): Released PipelinedSubpartition#1 [number of buffers: 0 (0 bytes), number of buffers in backlog: 0, finished? false, read view? false].
2019-01-02 03:03:47.548 [thread==> Source: Custom Source (1/1)] partition.ResultPartitionManager#releasePartitionsProducedBy:105 Released all partitions produced by 74a4ed4bb2f80aa2b98e11bd09ea64ef.
2019-01-02 03:03:47.548 [thread==> Source: Custom Source (1/1)] network.TaskEventDispatcher#unregisterPartition:78 unregistering 85c7415bae559d6198b8bb69d4c6e49f@74a4ed4bb2f80aa2b98e11bd09ea64ef
2019-01-02 03:03:47.548 [thread==> Source: Custom Source (1/1)] taskmanager.Task#run:919 Ensuring all FileSystem streams are closed for task Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) [CANCELED]
2019-01-02 03:03:47.549 [thread==> flink-akka.actor.default-dispatcher-2] taskexecutor.TaskExecutor#unregisterTaskAndNotifyFinalState:1337 Un-registering task and sending final execution state CANCELED to JobManager for task Source: Custom Source 74a4ed4bb2f80aa2b98e11bd09ea64ef.
2019-01-02 03:03:47.562 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#releaseSlot:745 Releasing slot [SlotRequestId{8a877431375df8aeadb2fd845cae15fc}] because: Slot is being returned to the SlotPool.
2019-01-02 03:03:47.566 [thread==> flink-akka.actor.default-dispatcher-3] executiongraph.Execution#transitionState:1316 Source: Custom Source (1/1) (74a4ed4bb2f80aa2b98e11bd09ea64ef) switched from CANCELING to CANCELED.
2019-01-02 03:03:47.566 [thread==> flink-akka.actor.default-dispatcher-3] executiongraph.Execution#processFail:1079 Ignoring transition of vertex Source: Custom Source (1/1) - execution #1 to FAILED while being CANCELED.
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#releaseSlot:745 Releasing slot [SlotRequestId{12bfcf1674f5b96567a076086dbbfd1b}] because: Slot is being returned to the SlotPool.
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] executiongraph.ExecutionGraph#tryRestartOrFail:1477 Try to restart or fail the job Flink Streaming Job (06e76f6e31728025b22fdda9fadd6f01) if no longer possible.
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] executiongraph.ExecutionGraph#transitionState:1356 Job Flink Streaming Job (06e76f6e31728025b22fdda9fadd6f01) switched from state FAILING to RESTARTING.
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] executiongraph.ExecutionGraph#tryRestartOrFail:1487 Restarting the job Flink Streaming Job (06e76f6e31728025b22fdda9fadd6f01).
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#releaseSlot:745 Releasing slot [SlotRequestId{6d7f0173c1d48e5559f6a14b080ee817}] because: Release multi task slot because all children have been released.
2019-01-02 03:03:47.567 [thread==> flink-akka.actor.default-dispatcher-3] slotpool.SlotPool#tryFulfillSlotRequestOrMakeAvailable:842 Adding returned slot [AllocationID{7b5faad9073d7fac6759e40981197b8d}] to available slots
2019-01-02 03:03:47.647 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:47.748 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:47.848 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:47.948 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.049 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.149 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.249 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.349 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.450 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
2019-01-02 03:03:48.550 [thread==> OutputFlusher for Source: Custom Source] io.StreamRecordWriter$OutputFlusher#run:192 Running OutputFlusher outputflusher1546369427544
{code}
 

 



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)