Good evening! I've downgraded Flink down to version 1.16.1 in order to use Gelly. I'm running a very simple app, which simply reads a large txt file (from hdfs) containing the edges of a graph, makes the graph, and then outputs the number of triangles in the graph (either by printing to console, or by writing it to a new file in hdfs). But, whenever I try to run the application on flink, where i have 1 job manager and 7 taskmanagers, one of them always fails at the first task, as shown in the screenshot below. The screenshot shows that there was a failure in the third job, and a cancellation in the first, but in fact, after looking at the error log for the taskmanager that failed the third task, the error states that the connection was unexpectedly closed by another taskmanager, which is always the one that didn't complete the first task.

(7c47ca8b3c53e4646c5f02c6e329a650_6b68f1bbaa75cb9b23c5012ad11acdd0_1_0) switched from RUNNING to FAILED with failure cause: java.lang.Exception: The data preparation for task 'CHAIN GroupReduce (GroupReduce at run(TriangleEnumerator.java:72)) -> Combine (Reduce at run(TriangleEnumerator.java:74))' , caused an error: java.util.concurrent.ExecutionException: java.lang.RuntimeException: Error obtaining the sorted input: Thread 'SortMerger Reading Thread' terminated due to an exception: Connection unexpectedly closed by remote task manager '10.0.1.72/10.0.1.72:45235 [ 10.0.1.72:35991-49e4c4 ] '. This might indicate that the remote task manager was lost.

I've been struggling to figure out what the cause of this might be. At first I thought it might be a problem with hadoop, but I tried a simpler program that reads the same data but just writes it to another file in hdfs, and it worked fine. I have the log from the task manager that didn't complete the first task, and it's set to loglevel=debug, because I couldn't find anything useful in info. I'm appending it here. I still think this might be related to hadoop because i'm noticing some weird things such as: • At the beginning, it states hadoop version = 2.8.3, which is not the case, my system is running hadoop 3.3.5, and I don't think I've set the version as an option somewhere. I've also placed the respective jar files in the lib folder, with hadoop version 3.3.5. • At various points, a WARN level message states Could not find Hadoop configuration via any of the supported methods (Flink configuration, environment variables). However, by looking at the source code, i can see that it should only display this message after displaying various debug messages, looking for the hadoop configuration in my system. This indicates that it can't even find the environmental variables that I've set, which is concerning. I appreciate any help I can get, and will gladly provide additional info if prompted.

One possible explanation is that the task manager is running out of memory while booting. 1. How large is the very large text file you mentioned? 2. How much memory have you configured for the task manager? 3. Are you loading the entire file into memory, or reading line by line?

The specific file is not that large. It's less than 300MB. I'm a little tight on memory, I have configured flink such that:

taskmanager.memory.flink.size: 6000m

I'm using one of gelly's graphbuilders to read the file and construct the graph instantly. I'm not aware of the specifics of how it goes about this, but I'd assume that it has built in optimizations for such cases:

Graph<LongValue, NullValue, NullValue> graph = Graph.fromCsvReader(hdfsFilePath, env).lineDelimiterEdges(" ").keyType(LongValue.class);

Wouldn't it display an error message if it run out of memory? Something like "JVM heap space" or something like this? Something definitely goes wrong with the node that doesn't complete the first task, because every time that I run the app, flink crashes on that VM, and I have to restart the cluster. But how would I go about figuring that out, and fixing it?

Normally I'd expect to see an OOM error. However, we've definitely seen some cases where the task manager memory fills up and crashes which leads to the

Connection unexpectedly closed by remote task manager

you mentioned above.

It doesn't sound impossible. What troubles me is that the log seems to end off abruptly. Not that I can understand exactly what's going on, mind you, but I would expect a message that says that the task was terminated, switching from running to cancelled, or something of the sort. Right now, the last two DEBUG messages simply state:

2023-07-13 14:12:45,876 DEBUG org.apache.flink.runtime.operators.DataSourceTask            [] - Starting to read input from split [0] <hdfs://10.0.1.70:9000/GraphXample/InputGraphs/graph_small.txt:0+40076936>:  CHAIN DataSource (org.apache.flink.graph.GraphCsvReader) -> Map (Type conversion) -> FlatMap (FlatMap at run(TriangleEnumerator.java:69)) (5/7)
2023-07-13 14:12:47,094 DEBUG org.apache.flink.runtime.taskexecutor.TaskExecutor           [] - Received heartbeat request from 4d66c7ae7a636b66f296fabcecffa9bc.

which for me, seems to indicate "working as normal". It is possible that I've configured the memory wrong. The number i've set as the taskmanager.memory.flink.size might be larger than my actual memory. Could that be an issue?

Maybe? I assume if you set a number larger than actual memory, the app would default to your actual memory limit. You mentioned this setting, is the unit here 6000 megs?

taskmanager.memory.flink.size: 6000m

I can't exactly figure out what the unit is, that why I put a number that I was sure was larger than my main memory. I decreased it to a number that I am sure is lower than my main memory, and I tried again. This time, I am getting a JVM Heap Space error. Do you think there's a way to combat this?

You can see what the value translates to in the task manager list of the flink dashboard.

Yeah I've checked that. It seems to be MB. Now that I've set the value to 4000m, I get this: 4000/1024 = 3.90625, which probably corresponds to Physical MEM.

Interesting. Is there a way you can split things up into smaller files? It seems that no matter what, the entire file contents are loaded into the

Graph

object, which will definitely exceed your memory limitations once the file is multiple gigs.