Apache Flink 1.20 Batch Job Failure (S3 Source to S3 Sink)

I am relatively new to AMS for Flink. I wrote a simple Flink 1.20 program using Java 11 to analyze a file in a S3 Bucket and write the results to another S3 Bucket. This program worked fine on a single nodelocal Linux filesystem but failed in AMS for Flink with the following error.

{ "applicationARN": "arn:aws:kinesisanalytics:us-east-2:047472788728:application/datahose-app", "applicationVersionId": "2", "locationInformation": "org.apache.flink.runtime.taskmanager.Task.transitionState(Task.java:1131)", "logger": "org.apache.flink.runtime.taskmanager.Task", "message": "Source: S3-Data-Source -> Flat Map (1/1)#0 (013a3701c2a1bb4a97998e5dae2e7b10_cbc357ccb763df2852fee8c4fc7d55f2_0_0) switched from RUNNING to FAILED with failure cause:", "messageSchemaVersion": "1", "messageType": "WARN", "threadName": "Source: S3-Data-Source -> Flat Map (1/1)#0", "throwableInformation": "java.lang.UnsupportedOperationException\n\tat org.apache.flink.runtime.io.network.partition.ResultPartition.getAllDataProcessedFuture(ResultPartition.java:231)\n\tat org.apache.flink.streaming.runtime.tasks.StreamTask.afterInvoke(StreamTask.java:988)\n\tat org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:923)\n\tat org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:972)\n\tat org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:951)\n\tat org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:765)\n\tat org.apache.flink.runtime.taskmanager.Task.run(Task.java:577)\n\tat java.base/java.lang.Thread.run(Thread.java:829)\n" }

I have pasted the a code snippet below from the main Java Class. Can somebody tell me what am I missing?

   LOG.info("S3 Input Path: {}", s3InputPath);
    LOG.info("S3 Output Path: {}", s3OutputPath);
    LOG.info("Region: {}", region);

    FileSource<String> fileSource = forRecordStreamFormat(new TextLineInputFormat(), new Path(s3InputPath)).build();
    DataStream<String> sourceRecords = env.fromSource(fileSource, WatermarkStrategy.noWatermarks(), "S3-Data-Source");

    // Parse the data to obtain a Tuple of (name,1) for each record
    DataStream<Tuple2<String, Integer>> parsedData = sourceRecords.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
        @Override
        public void flatMap(String record, Collector<Tuple2<String, Integer>> out) {
               String[] elements = record.trim().split(",");
               if(elements.length ==2) {
                   // Emit (name,1) for each record
                   out.collect(new Tuple2<>(elements[0].trim(), 1));
               }
        }
    });

    // Group by name and sum the counts
    DataStream<Tuple2<String, Integer>> visitsPerPerson = parsedData.keyBy(value ->  value.f0).sum(1);

    // Transform to the Output Map
    DataStream<String> results = visitsPerPerson.map(new MapFunction<Tuple2<String, Integer>, String>() {
        @Override
        public String map(Tuple2<String, Integer> value) {
            return value.f0 + " visited the gym " + value.f1 + " times";
        }
    });

    FileSink<String> s3Sink = FileSink.forRowFormat(new Path(s3OutputPath), new SimpleStringEncoder<String>("UTF-8"))               .withRollingPolicy(DefaultRollingPolicy.builder().withRolloverInterval(Duration.ofSeconds(5)).withInactivityInterval(Duration.ofSeconds(3)).build())
            .build();

    results.sinkTo(s3Sink);
    env.execute("S3 to S3 - Analytical Use Case");

Thanks!!