EMR Fink Jobs Log Agreegation, Rolling and shifting to S3

I am running Flink jobs in the AWS EMR. I current as-is situation is,

1. The flink job logs are currently getting stored in HDFS and it grows over time.
2. Currently there is no log rolling happening
3. The logs are available in S3 after some point of time, but it is not getting cleaned up from HDFS.

So i did the following,

I added the following configuration as part of software configuration while creating the cluster.

{ "Classification": "yarn-site", "Properties": { "yarn.log-aggregation-enable": "true", "yarn.log-aggregation.retain-seconds": "-1", "yarn.nodemanager.log-aggregation.roll-monitoring-interval-seconds": "3600" } }

then i see, the agreegation is happening, but i am not seeing any rolling files. I expected for every 1 hour, i will get a log file taskmanager.log.gz and then taskmanager.1.log.gz, taskmanager.2.log.gz in S3 but it is not the case. I always see only one file taskmanager.log.gz

My second clarification, when i added the below configuration, there are few folders getting created like this hadoop_$folder$, bucket-logs-tfile_$folder$ and no logs are getting created..

yarn.nodemanager.remote-app-log-dir: s3://<bukcetname>/EMRLogs

My third clarification is, do i need to add the following flink-log4j configuration? Does it have any impact on the Rolling?

{ "Classification": "flink-log4j", "Properties": { "appender.main.fileName": "${sys:log.file}", "appender.main.filePattern": "${sys:log.file}.%i", "appender.main.layout.pattern": "%d{yyyy-MM-dd HH:mm:ss} %-5p %m%n", "appender.main.layout.type": "PatternLayout", "appender.main.name": "RollingFile", "appender.main.policies.size.size": "10KB", "appender.main.policies.size.type": "SizeBasedTriggeringPolicy", "appender.main.policies.type": "Policies", "appender.main.strategy.max": "5", "appender.main.strategy.type": "DefaultRolloverStrategy", "appender.main.type": "RollingFile", "logger.AbstractStreamingCommitterHandler.level": "WARN", "logger.AbstractStreamingCommitterHandler.name": "org.apache.flink.streaming.runtime.operators.sink.AbstractStreamingCommitterHandler", "logger.CheckpointCoordinator.level": "WARN", "logger.CheckpointCoordinator.name": "org.apache.flink.runtime.checkpoint.CheckpointCoordinator", "logger.SourceCoordinator.level": "WARN", "logger.SourceCoordinator.name": "org.apache.flink.runtime.source.coordinator.SourceCoordinator" }