title: Flink S3 StreamingFileSink not writing files to S3
body: <p>I am doing a POC for writing data to S3 using Flink. The program does not give a error. However I do not see any files being written in S3 either.</p>
  <p>Below is the code</p>
  <pre class="lang-java prettyprint-override"><code>
  public class StreamingJob {
  
      public static void main(String[] args) throws Exception {
          // set up the streaming execution environment
          final String outputPath = &quot;s3a://testbucket-s3-flink/data/&quot;;
          final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
          
          //Enable checkpointing
          env.enableCheckpointing();
          //S3 Sink
          final StreamingFileSink&lt;String&gt; sink = StreamingFileSink
                  .forRowFormat(new Path(outputPath), new SimpleStringEncoder&lt;String&gt;(&quot;UTF-8&quot;))
                  .build();
  
  
          //Source is a local kafka
          Properties properties = new Properties();
          properties.setProperty(&quot;bootstrap.servers&quot;, &quot;kafka:9094&quot;);
          properties.setProperty(&quot;group.id&quot;, &quot;test&quot;);
          
  
          DataStream&lt;String&gt; input = env.addSource(new FlinkKafkaConsumer&lt;String&gt;(&quot;queueing.transactions&quot;, new SimpleStringSchema(), properties));
          
          
          input.flatMap(new Tokenizer()) // Tokenizer for generating words
                  .keyBy(0) // Logically partition the stream for each word
                  .timeWindow(Time.minutes(1)) // Tumbling window definition
                  .sum(1) // Sum the number of words per partition
                  .map(value -&gt; value.f0 + &quot; count: &quot; + value.f1.toString() + &quot;\n&quot;)
                  .addSink(sink);
  
          // execute program
          env.execute(&quot;Flink Streaming Java API Skeleton&quot;);
      }
  
      public static final class Tokenizer
              implements FlatMapFunction&lt;String, Tuple2&lt;String, Integer&gt;&gt; {
  
          @Override
          public void flatMap(String value, Collector&lt;Tuple2&lt;String, Integer&gt;&gt; out) {
              String[] tokens = value.toLowerCase().split(&quot;\\W+&quot;);
              for (String token : tokens) {
                  if (token.length() &gt; 0) {
                      out.collect(new Tuple2&lt;&gt;(token, 1));
                  }
              }
          }
      }
  }
  
  </code></pre>
  <p>Note that I have set the <code>s3.access-key</code> and <code>s3.secret-key</code> value in the configuration and tested by changing them to incorrect values (I got a error on incorrect values)</p>
  <p>Any pointers what may be going wrong?</p>