title: Flink StreamingFileSink to Amazon S3 with flink-s3-fs-hadoop token expired exception
body: <p>I'm trying to stream some data from kafka into s3 (using s3a protocol).</p>
  <p>The pipeline works good for an hour, but after an hour (same as my token expiry setting for AWS), throws a (from the StreamingFileSink):</p>
  <pre><code>Caused by: com.amazonaws.services.s3.model.AmazonS3Exception: The provided token has expired. (Service: Amazon S3; Status Code: 400; Error Code: ExpiredToken; Request ID: 7YFGVQ92YT51DP0K; S3 Extended Request ID: sx6UJJ548o0wpwJbkoWJ16jKRVih3ZV9XQdbThNhq5kUU7A7yCx58tcCGELVs5tqGWaMMPfZxZM=; Proxy: webproxy)
      at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleErrorResponse(AmazonHttpClient.java:1819)
      at com.amazonaws.http.AmazonHttpClient$RequestExecutor.handleServiceErrorResponse(AmazonHttpClient.java:1403)
      at com.amazonaws.http.AmazonHttpClient$RequestExecutor.executeOneRequest(AmazonHttpClient.java:1372)
  ...
  </code></pre>
  <p>I'm using AWSCredentialsProvider implementation which implements the <code>getCredentials</code> and refreshes the token every 15 minutes with a newly parsed secret from aws.</p>
  <p>My assumption is that the issue lies in how I initialise the <code>StreamingFileSink</code> in the job itself:</p>
  <pre class="lang-java prettyprint-override"><code> StreamExecutionEnvironment env = getStreamExecutionEnvironment();
  
   StreamingFileSink&lt;FELEvent&gt; sink = StreamingFileSink
                  .forBulkFormat(&lt;my Path Settings with basePath s3a://bucket/path/to/dir&gt;)
                  .withRollingPolicy(OnCheckpointRollingPolicy.build())
                  .withNewBucketAssigner(&lt;My custom bucket assigner&gt;)
                  .build();
  
  
  env.fromSource(&lt;Kafka source&gt;)
                  .map(&lt;Some operation&gt;)
                  .filter(&lt;Some filtergin&gt;)
                  .addSink(sink)
                  .name(&quot;name&quot;).uid(&quot;uid&quot;);
  
  env.execute(&quot;TAG&quot;);
  
  
  </code></pre>
  <p>Any ideas if the plugin refreshes tokens for an already initialised StreamingFileSink?
  If no, what's the best way to handle this scenario?</p>
  <p>(Because of compatibility issues with zookeeper I'm using flink 14.3.)</p>