Hello flink community, I am working on a flink job to consume events from kafka as source and write the data stream events to S3 bucket (s3 as sink basically). Here’s the sample code for the same. I have the S3 credentials configured in the flink-conf.yml. The checkpointing works well and I can see the metadata files being written to S3. But the events from data stream are not being written to the S3 buckets and there’s no error in the sink while the job is running. I am unable to figure out what’s going wrong ! Any help on this pls ! Thanks

public void run() throws Exception {
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
        environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        environment.setStateBackend(new FsStateBackend("<s3://dev/checkpoint>"));
        environment.enableCheckpointing(50000L);
        environment.getConfig().setGlobalJobParameters(parameters);

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "temp");

        DataStream<ObjectNode> source = environment
                .addSource(new FlinkKafkaConsumer<>("test", new JSONKeyValueDeserializationSchema(false), properties))
                .rebalance();

        SingleOutputStreamOperator<Tuple4<String, String, Integer, Long>> workflow = source
                .map(new RecordMapper())
                .keyBy(0, 1)
                .timeWindow(Time.minutes(2))
                .aggregate(new TimeStampAggregateFunction());

        //S3 sink
        StreamingFileSink<Tuple4<String, String, Integer, Long>> sink = StreamingFileSink
                .forRowFormat(new Path("<s3://dev/sink-test>"),
                        (Encoder<Tuple4<String, String, Integer, Long>>) (element, stream) -> {
                            PrintStream out = new PrintStream(stream);
                            out.println(element.f1);
                        })
                .withBucketAssigner(new BucketAssigner())
                .build();

        workflow.addSink(sink);

        environment.execute();
    }

Hi Did you get this to work?