[FLINK-34616] python dist doesn't clean when open method construct resource - ASF JIRA

Details

Type: Bug
Status: Closed
Priority: Major
Resolution: Fixed
Affects Version/s: 1.19.0, 1.18.1
Fix Version/s: 1.18.2, 1.20.0, 1.19.1
Component/s: API / Python
Labels:
- pull-request-available

Description

our enviroment found lots of python-dist causing disk full.

the main resource is
constructEnvironmentVariables -> constructArchivesDirectory -> CompressionUtils.extractFile which has
ClosedByInterruptException Exception and the root exception has lost. we found it by arthas.
and it will not run the clean dir logic

2024-03-07 18:19:34,265 ERROR [[vertex-1]MiniBatchAssigner(interval=[5000ms], mode=[ProcTime]) -> PythonCalc(select=content, sourc (18/128)#31 org.apache.flink.python.env.AbstractPythonEnvironmentManager [] - Error when create resource.
java.nio.channels.ClosedByInterruptException: null
at java.nio.channels.spi.AbstractInterruptibleChannel.end(AbstractInterruptibleChannel.java:199) ~[?:?]
at sun.nio.ch.FileChannelImpl.endBlocking(FileChannelImpl.java:162) ~[?:?]
at sun.nio.ch.FileChannelImpl.readInternal(FileChannelImpl.java:816) ~[?:?]
at sun.nio.ch.FileChannelImpl.read(FileChannelImpl.java:796) ~[?:?]
at org.apache.commons.compress.archivers.zip.ZipFile$BoundedFileChannelInputStream.read(ZipFile.java:1420) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.commons.compress.utils.BoundedArchiveInputStream.read(BoundedArchiveInputStream.java:82) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at java.io.BufferedInputStream.fill(BufferedInputStream.java:252) ~[?:?]
at java.io.BufferedInputStream.read1(BufferedInputStream.java:292) ~[?:?]
at java.io.BufferedInputStream.read(BufferedInputStream.java:351) ~[?:?]
at java.io.SequenceInputStream.read(SequenceInputStream.java:199) ~[?:?]
at java.util.zip.InflaterInputStream.fill(InflaterInputStream.java:243) ~[?:?]
at org.apache.commons.compress.archivers.zip.InflaterInputStreamWithStatistics.fill(InflaterInputStreamWithStatistics.java:52) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at java.util.zip.InflaterInputStream.read(InflaterInputStream.java:159) ~[?:?]
at org.apache.commons.compress.archivers.zip.InflaterInputStreamWithStatistics.read(InflaterInputStreamWithStatistics.java:67) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at java.io.FilterInputStream.read(FilterInputStream.java:107) ~[?:?]
at org.apache.flink.util.IOUtils.copyBytes(IOUtils.java:61) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.util.IOUtils.copyBytes(IOUtils.java:86) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.util.CompressionUtils.extractZipFileWithPermissions(CompressionUtils.java:223) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.util.CompressionUtils.extractFile(CompressionUtils.java:61) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager.constructArchivesDirectory(AbstractPythonEnvironmentManager.java:365) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager.constructEnvironmentVariables(AbstractPythonEnvironmentManager.java:178) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager.lambda$open$0(AbstractPythonEnvironmentManager.java:126) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager$PythonEnvResources.createResource(AbstractPythonEnvironmentManager.java:468) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager$PythonEnvResources.getOrAllocateSharedResource(AbstractPythonEnvironmentManager.java:435) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.python.env.AbstractPythonEnvironmentManager.open(AbstractPythonEnvironmentManager.java:114) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.api.operators.python.AbstractEmbeddedPythonFunctionOperator.open(AbstractEmbeddedPythonFunctionOperator.java:68) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.table.runtime.operators.python.scalar.EmbeddedPythonScalarFunctionOperator.open(EmbeddedPythonScalarFunctionOperator.java:127) ~[flink-python_2.12-1.15.2-SNAPSHOT-jar-with-dependencies.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.runtime.tasks.RegularOperatorChain.initializeStateAndOpenOperators(RegularOperatorChain.java:107) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreGates(StreamTask.java:703) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.runtime.tasks.StreamTaskActionExecutor$1.call(StreamTaskActionExecutor.java:55) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restoreInternal(StreamTask.java:679) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.streaming.runtime.tasks.StreamTask.restore(StreamTask.java:646) ~[flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:963) [flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:932) [flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:756) [flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:578) [flink-dist_2.12-1.15.2-SNAPSHOT.jar:1.15.2-SNAPSHOT]
at java.lang.Thread.run(Thread.java:991) [?:?]