Re: Re:Flink batch job memory/disk leak when invoking set method on a static Configuration object.

[Vadim Vararu]

Hi,

I've run it on a standalone Flink cluster. No Yarn involved.
________________________________
From: Haibo Sun <sunhaib...@163.com>
Sent: Friday, June 28, 2019 6:13 AM
To: Vadim Vararu
Cc: user@flink.apache.org
Subject: Re:Flink batch job memory/disk leak when invoking set method on a 
static Configuration object.

Hi, Vadim

This similar issue has occurred in earlier versions, see 
https://issues.apache.org/jira/browse/FLINK-4485.
Are you running a Flink session cluster on YARN? I think it might be a bug. 
I'll see if I can reproduce it with the master branch code, and if yes, I will 
try to investigate it.

If someone already knows the cause of the problem, that's the best,  it won't 
need to be re-investigated.

Best,
Haibo


At 2019-06-28 00:46:43, "Vadim Vararu" <vadim.var...@adswizz.com> wrote:
Hi guys,

I have a simple batch job with a custom output formatter that writes to a local 
file.


public class JobHadoop {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = 
ExecutionEnvironment.getExecutionEnvironment();

        env.fromCollection(Sets.newHashSet("line1", "line2", "line3"))
                .map(line -> line + "dd")
                .write(new HadoopUsageOutputFormat(), "file:///tmp/out");

        env.execute(JobHadoop.class.getName());
    }

}

public class HadoopUsageOutputFormat extends FileOutputFormat<String> 
implements OutputFormat<String> {

    private static final Configuration DEFAULT_HADOOP_CONFIGURATION = new 
Configuration();
    public static final String DEFAULT_LINE_DELIMITER = "\n";

    private Writer writer;

    static {
        DEFAULT_HADOOP_CONFIGURATION.set("just.a.prop", "/var/test1");
    }

    @Override
    public void open(int taskNumber, int numTasks) throws IOException {
        super.open(taskNumber, numTasks);
        writer = new OutputStreamWriter(new BufferedOutputStream(stream));
    }

    @Override
    public void writeRecord(String record) throws IOException {
        writer.write(record);
        writer.write(DEFAULT_LINE_DELIMITER);
    }

    @Override
    public void close() throws IOException {
        if (writer != null) {
            this.writer.flush();
            this.writer.close();
        }

        super.close();
    }
}

The problem is that after the job is finished, there is somewhere a memory leak 
that does not permit the blobStore of the job to be deleted. The number of such 
"deleted" files increases after each job run. Even if they are marked as 
deleted, there is somewhere a reference to the file in the JobManager process 
that keeps it from actual deletion.

[cid:86fb55b$1$16b9c126cfc$Coremail$sunhaibotb$163.com]


Also, the problem reproduces only if I actually invoke the set method of 
Configuration:

static {
        DEFAULT_HADOOP_CONFIGURATION.set("just.a.prop", "/var/test1");
    }

>From my observations, if I change the

private static final Configuration DEFAULT_HADOOP_CONFIGURATION = new 
Configuration();

to a non-static field, then the problem does no reproduce any more.


However, I'm interested if that's a normal behaviour or a bug/leak somewhere in 
Flink itself.

Thanks,
Vadim.