[jira] [Created] (FLINK-32809) YarnClusterDescriptor.isarchiveonlyincludedinShipArchiveFiles dose not work as expected
junzhong qin created FLINK-32809: Summary: YarnClusterDescriptor.isarchiveonlyincludedinShipArchiveFiles dose not work as expected Key: FLINK-32809 URL: https://issues.apache.org/jira/browse/FLINK-32809 Project: Flink Issue Type: Bug Components: Deployment / YARN Affects Versions: 1.17.1, 1.16.2, 1.18.0 Reporter: junzhong qin YarnClusterDescriptor.isarchiveonlyincludedinShipArchiveFiles(List shipFiles) check wether the shipFiles are all archive files, but it dose not work as expected. {code:java} public static boolean isArchiveOnlyIncludedInShipArchiveFiles(List shipFiles) { return shipFiles.stream() .filter(File::isFile) .map(File::getName) .map(String::toLowerCase) .allMatch( name -> name.endsWith(".tar.gz") || name.endsWith(".tar") || name.endsWith(".tgz") || name.endsWith(".dst") || name.endsWith(".jar") || name.endsWith(".zip")); } {code} When we pass a directory and an archive file it should return false but it returns true. {code:java} // dir1 is a directory and archive.zip is an archive file List files = Arrays.asList(new File("/tmp/dir1"), new File("/tmp/archive.zip")); boolean result = isArchiveOnlyIncludedInShipArchiveFiles(files); System.out.println(result); // Print true but is should print false{code} If flink user want to add directory as ship file, they should use configuration: yarn.ship-files . -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-31749) The Using Hadoop OutputFormats example is not avaliable for DataStream
junzhong qin created FLINK-31749: Summary: The Using Hadoop OutputFormats example is not avaliable for DataStream Key: FLINK-31749 URL: https://issues.apache.org/jira/browse/FLINK-31749 Project: Flink Issue Type: Improvement Components: Documentation Affects Versions: 1.15.4, 1.17.0 Reporter: junzhong qin The following example shows how to use Hadoop’s {{TextOutputFormat from the doc: https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/connectors/datastream/formats/hadoop/#using-hadoop-outputformats . But the DataStream has no outpu()).}} {code:java} // Obtain the result we want to emit DataStream> hadoopResult = [...] // Set up the Hadoop TextOutputFormat. HadoopOutputFormat hadoopOF = // create the Flink wrapper. new HadoopOutputFormat( // set the Hadoop OutputFormat and specify the job. new TextOutputFormat(), job ); hadoopOF.getConfiguration().set("mapreduce.output.textoutputformat.separator", " "); TextOutputFormat.setOutputPath(job, new Path(outputPath)); // Emit data using the Hadoop TextOutputFormat. hadoopResult.output(hadoopOF); {code} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-33679) RestoreMode uses NO_CLAIM as default instead of LEGACY
junzhong qin created FLINK-33679: Summary: RestoreMode uses NO_CLAIM as default instead of LEGACY Key: FLINK-33679 URL: https://issues.apache.org/jira/browse/FLINK-33679 Project: Flink Issue Type: Improvement Components: Documentation, Runtime / State Backends Reporter: junzhong qin RestoreMode uses NO_CLAIM as default instead of LEGACY. {code:java} public enum RestoreMode implements DescribedEnum { CLAIM( "Flink will take ownership of the given snapshot. It will clean the" + " snapshot once it is subsumed by newer ones."), NO_CLAIM( "Flink will not claim ownership of the snapshot files. However it will make sure it" + " does not depend on any artefacts from the restored snapshot. In order to do that," + " Flink will take the first checkpoint as a full one, which means it might" + " reupload/duplicate files that are part of the restored checkpoint."), LEGACY( "This is the mode in which Flink worked so far. It will not claim ownership of the" + " snapshot and will not delete the files. However, it can directly depend on" + " the existence of the files of the restored checkpoint. It might not be safe" + " to delete checkpoints that were restored in legacy mode "); private final String description; RestoreMode(String description) { this.description = description; } @Override @Internal public InlineElement getDescription() { return text(description); } public static final RestoreMode DEFAULT = NO_CLAIM; } {code} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-34261) When slotmanager.redundant-taskmanager-num is set to a value greater than 1, redundant task managers may be repeatedly released and requested
junzhong qin created FLINK-34261: Summary: When slotmanager.redundant-taskmanager-num is set to a value greater than 1, redundant task managers may be repeatedly released and requested Key: FLINK-34261 URL: https://issues.apache.org/jira/browse/FLINK-34261 Project: Flink Issue Type: Bug Reporter: junzhong qin Attachments: image-2024-01-29-17-29-15-453.png Redundant task managers are extra task managers started by Flink, to speed up job recovery in case of failures due to task manager lost. But when we configured {code:java} slotmanager.redundant-taskmanager-num: 2 // any value greater than 1{code} Flink will release and request redundant TM repeatedly. We can reproduce this situation by using [Flink Kubernetes Operator (using minikube here)|https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/], here is an example yaml file: {code:java} // redundant-tm.yaml # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. apiVersion: flink.apache.org/v1beta1 kind: FlinkDeployment metadata: name: redundant-tm spec: image: flink:1.18 flinkVersion: v1_18 flinkConfiguration: taskmanager.numberOfTaskSlots: "2" slotmanager.redundant-taskmanager-num: "2" cluster.fine-grained-resource-management.enabled: "false" serviceAccount: flink jobManager: resource: memory: "1024m" cpu: 1 taskManager: resource: memory: "1024m" cpu: 1 job: jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar parallelism: 3 upgradeMode: stateless logConfiguration: log4j-console.properties: |+ rootLogger.level = DEBUG rootLogger.appenderRef.file.ref = LogFile rootLogger.appenderRef.console.ref = LogConsole appender.file.name = LogFile appender.file.type = File appender.file.append = false appender.file.fileName = ${sys:log.file} appender.file.layout.type = PatternLayout appender.file.layout.pattern = %d{-MM-dd HH:mm:ss,SSS} %-5p %-60c %x - %m%n{code} After executing: {code:java} kubectl create -f redundant-tm.yaml kubectl port-forward svc/redundant-tm 8081{code} We can find repeatedly release and request redundant TM in JM's log: {code:java} 2024-01-29 09:26:25,033 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Registering TaskManager with ResourceID redundant-tm-taskmanager-1-4 (pekko.tcp://flink@10.244.1.196:6122/user/rpc/taskmanager_0) at ResourceManager2024-01-29 09:26:25,060 DEBUG org.apache.flink.runtime.resourcemanager.slotmanager.DeclarativeSlotManager [] - Registering task executor redundant-tm-taskmanager-1-4 under 44c649b2d84e87cdd5e6c53971f8b877 at the slot manager.2024-01-29 09:26:25,061 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Worker redundant-tm-taskmanager-1-4 is registered.2024-01-29 09:26:25,061 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Worker redundant-tm-taskmanager-1-4 with resource spec WorkerResourceSpec {cpuCores=1.0, taskHeapSize=25.600mb (26843542 bytes), taskOffHeapSize=0 bytes, networkMemSize=64.000mb (67108864 bytes), managedMemSize=230.400mb (241591914 bytes), numSlots=2} was requested in current attempt. Current pending count after registering: 1.2024-01-29 09:26:25,196 DEBUG org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Update resource declarations to [ResourceDeclaration{spec=WorkerResourceSpec {cpuCores=1.0, taskHeapSize=25.600mb (26843542 bytes), taskOffHeapSize=0 bytes, networkMemSize=64.000mb (67108864 bytes), managedMemSize=230.400mb (241591914 bytes), numSlots=2}, numNeeded=3, unwantedWorkers=[]}].2024-01-29 09:26:25,196 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - need release 1 workers, current worker number 4, declared worker number 32024-01-29 09:26:25,199 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResour
[jira] [Created] (FLINK-34525) When redundant task managers are enabled, it will still take a considerable amount of time to restart a job if a task manager is killed.
junzhong qin created FLINK-34525: Summary: When redundant task managers are enabled, it will still take a considerable amount of time to restart a job if a task manager is killed. Key: FLINK-34525 URL: https://issues.apache.org/jira/browse/FLINK-34525 Project: Flink Issue Type: Improvement Reporter: junzhong qin -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-34526) Without shuffling data, when a Task Manager is killed, restarting the Flink job takes a considerable amount of time
junzhong qin created FLINK-34526: Summary: Without shuffling data, when a Task Manager is killed, restarting the Flink job takes a considerable amount of time Key: FLINK-34526 URL: https://issues.apache.org/jira/browse/FLINK-34526 Project: Flink Issue Type: Sub-task Reporter: junzhong qin Attachments: image-2024-02-27-15-50-25-071.png, image-2024-02-27-15-50-39-337.png In our test case, the pipeline is: !image-2024-02-27-15-50-25-071.png! parallelism = 100 and taskmanager.numberOfTaskSlots = 2 When the worker was killed at 2024-02-27 15:10:13,691 {code:java} 2024-02-27 15:10:13,691 INFO org.apache.flink.runtime.resourcemanager.active.ActiveResourceManager [] - Worker container_e2472_1706081484717_60538_01_50 is terminated. Diagnostics: Container container_e2472_1706081484717_60538_01_50 marked as failed. Exit code:137. Diagnostics:[2024-02-27 15:10:12.720]Container killed on request. Exit code is 137[2024-02-27 15:10:12.763]Container exited with a non-zero exit code 137. [2024-02-27 15:10:12.839]Killed by external signal {code} It took about 20 seconds to restart the job. {code:java} 2024-02-27 15:10:30,749 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: datagen_source[1] -> Sink: print_sink[2] (70/100) (2a1d06e6610bc499475fa6e647f8cac9_d3f21cabc6fe0fdf76c8be915bdb22a2_69_0) switched from RUNNING to FAILED on container_e2472_1706081484717_60538_01_50 @ ip-10-169-18-98.idata-server.shopee.io (dataPort=38597).org.apache.flink.runtime.jobmaster.JobMasterException: TaskManager with id container_e2472_1706081484717_60538_01_50(ip-10-169-18-98.idata-server.shopee.io:5454) is no longer reachable.at org.apache.flink.runtime.jobmaster.JobMaster$TaskManagerHeartbeatListener.notifyTargetUnreachable(JobMaster.java:1515) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT]at org.apache.flink.runtime.heartbeat.DefaultHeartbeatMonitor.reportHeartbeatRpcFailure(DefaultHeartbeatMonitor.java:126) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT]at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl.runIfHeartbeatMonitorExists(HeartbeatManagerImpl.java:275) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT]at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl.reportHeartbeatTargetUnreachable(HeartbeatManagerImpl.java:267) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT]at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl.handleHeartbeatRpcFailure(HeartbeatManagerImpl.java:262) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT]at org.apache.flink.runtime.heartbeat.HeartbeatManagerImpl.lambda$handleHeartbeatRpc$0(HeartbeatManagerImpl.java:248) ~[flink-dist-shopee-1.17-SNAPSHOT.jar:shopee-1.17-SNAPSHOT] // Deploy and run task 2024-02-27 15:10:32,426 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: datagen_source[1] -> Sink: print_sink[2] (70/100) (2a1d06e6610bc499475fa6e647f8cac9_d3f21cabc6fe0fdf76c8be915bdb22a2_69_1) switched from DEPLOYING to INITIALIZING.2024-02-27 15:10:32,427 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: datagen_source[1] -> Sink: print_sink[2] (69/100) (2a1d06e6610bc499475fa6e647f8cac9_d3f21cabc6fe0fdf76c8be915bdb22a2_68_1) switched from DEPLOYING to INITIALIZING.2024-02-27 15:10:33,347 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: datagen_source[1] -> Sink: print_sink[2] (70/100) (2a1d06e6610bc499475fa6e647f8cac9_d3f21cabc6fe0fdf76c8be915bdb22a2_69_1) switched from INITIALIZING to RUNNING.2024-02-27 15:10:33,421 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Source: datagen_source[1] -> Sink: print_sink[2] (69/100) (2a1d06e6610bc499475fa6e647f8cac9_d3f21cabc6fe0fdf76c8be915bdb22a2_68_1) switched from INITIALIZING to RUNNING. {code} -- This message was sent by Atlassian Jira (v8.20.10#820010)
[jira] [Created] (FLINK-34528) With shuffling data, when a Task Manager is killed, restarting the Flink job takes a considerable amount of time
junzhong qin created FLINK-34528: Summary: With shuffling data, when a Task Manager is killed, restarting the Flink job takes a considerable amount of time Key: FLINK-34528 URL: https://issues.apache.org/jira/browse/FLINK-34528 Project: Flink Issue Type: Sub-task Reporter: junzhong qin Attachments: image-2024-02-27-16-35-04-464.png In the test case, the pipeline looks like: !image-2024-02-27-16-35-04-464.png! The Source: Custom Source generates strings, and the job keyBy the strings to Sink: Unnamed. # parallelism = 100 # taskmanager.numberOfTaskSlots = 2 # disable checkpoint The worker was killed at {code:java} 2024-02-27 16:41:49,982 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Sink: Unnamed (6/100) (2f1c7b22098a273f5471e3e8f794e1d3_0a448493b4782967b150582570326227_5_0) switched from RUNNING to FAILED on container_e2472_1705993319725_62292_01_46 @ xxx (dataPort=38827).org.apache.flink.runtime.io.network.netty.exception.RemoteTransportException: Connection unexpectedly closed by remote task manager 'xxx/10.169.18.138:35983 [ container_e2472_1705993319725_62292_01_10(xxx:5454) ] '. This might indicate that the remote task manager was lost.at org.apache.flink.runtime.io.network.netty.CreditBasedPartitionRequestClientHandler.channelInactive(CreditBasedPartitionRequestClientHandler.java:134) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241) at org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter.channelInactive(ChannelInboundHandlerAdapter.java:81) at org.apache.flink.runtime.io.network.netty.NettyMessageClientDecoderDelegate.channelInactive(NettyMessageClientDecoderDelegate.java:94) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.fireChannelInactive(AbstractChannelHandlerContext.java:241) at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline$HeadContext.channelInactive(DefaultChannelPipeline.java:1405) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:262) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannelHandlerContext.invokeChannelInactive(AbstractChannelHandlerContext.java:248) at org.apache.flink.shaded.netty4.io.netty.channel.DefaultChannelPipeline.fireChannelInactive(DefaultChannelPipeline.java:901) at org.apache.flink.shaded.netty4.io.netty.channel.AbstractChannel$AbstractUnsafe$7.run(AbstractChannel.java:813) at org.apache.flink.shaded.netty4.io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174) at org.apache.flink.shaded.netty4.io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167) at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470) at org.apache.flink.shaded.netty4.io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:403) at org.apache.flink.shaded.netty4.io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997) at org.apache.flink.shaded.netty4.io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) at java.lang.Thread.run(Thread.java:748) ~[?:1.8.0_252]{code} The job took about 16 seconds to restart. {code:java} // The task was scheduled to a task manager that had already been killed 2024-02-27 16:41:53,506 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Deploying Source: Custom Source (16/100) (attempt #3) with attempt id 2f1c7b22098a273f5471e3e8f794e1d3_bc764cd8ddf7a0cff126f51c16239658_15_3 and vertex id bc764cd8ddf7a0cff126f51c16239658_15 to container_e2472_1705993319725_62292_01_10 @ xxx (dataPort=35983) with allocation id 975dded4548ad15b36d0e5e6aac8f5b6 // The last task switched from INITIALIZING to RUNNING 2024-02-27 16:42:05,176 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph [] - Sink: Unnamed (64/100) (2f1c7b22098a273f5471e3e8f794e1d3_0a448493b4782967b150582570326227_63_14) switched from INITIALIZING to RUNNING. {code} -- Th