title: kafka-connect Multipart uploads to s3 with s3rver
body: <p>I am trying to have a full local kafka stack with docker-compose and all my services are up and running (zookeeper, kafka broker, kafka-connect with s3 sink connector etc.)</p>
  <p>The producer outputs every second a simple avro object that I want to store in a s3 mock (s3rver). This is how I instantiate the kafka-connect:</p>
  <pre><code>curl -i -X PUT -H &quot;Accept:application/json&quot; \
      -H  &quot;Content-Type:application/json&quot; http://localhost:8083/connectors/sink-s3-test/config \
      -d '
   {
          &quot;connector.class&quot;: &quot;io.confluent.connect.s3.S3SinkConnector&quot;,
          &quot;key.converter&quot;:&quot;org.apache.kafka.connect.storage.StringConverter&quot;,
          &quot;tasks.max&quot;: &quot;1&quot;,
          &quot;topics&quot;: &quot;test&quot;,
          &quot;store.url&quot;: &quot;http://s3rver:8000&quot;,
          &quot;s3.region&quot;: &quot;eu-west-1&quot;,
          &quot;s3.bucket.name&quot;: &quot;test&quot;,
          &quot;flush.size&quot;: &quot;1&quot;,
          &quot;storage.class&quot;: &quot;io.confluent.connect.s3.storage.S3Storage&quot;,
          &quot;format.class&quot;: &quot;io.confluent.connect.s3.format.parquet.ParquetFormat&quot;,
          &quot;schema.generator.class&quot;: &quot;io.confluent.connect.storage.hive.schema.DefaultSchemaGenerator&quot;,
          &quot;schema.compatibility&quot;: &quot;NONE&quot;,
          &quot;partitioner.class&quot;: &quot;io.confluent.connect.storage.partitioner.DefaultPartitioner&quot;
          }
  '
  </code></pre>
  <p>But I get the following error:</p>
  <pre><code>kafka-producer     | [main] INFO ProducerLogger - test-0@947 written to partition 0
  kafka-producer     | [main] INFO ProducerLogger - sleep for one second...
  kafka-connect      | [2021-02-11 10:53:10,790] INFO Opening record writer for: topics/test/partition=0/test+0+0000000000.snappy.parquet (io.confluent.connect.s3.format.parquet.ParquetRecordWriterProvider)
  kafka-connect      | [2021-02-11 10:53:10,798] INFO Starting commit and rotation for topic partition test-0 with start offset {partition=0=0} (io.confluent.connect.s3.TopicPartitionWriter)
  kafka-connect      | [2021-02-11 10:53:10,798] INFO Flushing mem columnStore to file. allocated memory: 56 (org.apache.parquet.hadoop.InternalParquetRecordWriter)
  s3rver_1           | info: [S3rver] POST /test/topics/test/partition%3D0/test%2B0%2B0000000000.snappy.parquet?uploads 200 253 - 1.167 ms
  s3rver_1           | info: [S3rver] Stored object &quot;4d7da2e6e912fd41d4547ef57a056ab3_1&quot; in bucket &quot;test&quot; successfully
  s3rver_1           | info: [S3rver] PUT /test/topics/test/partition%3D0/test%2B0%2B0000000000.snappy.parquet?uploadId=4d7da2e6e912fd41d4547ef57a056ab3&amp;partNumber=1 200 - - 2.631 ms
  kafka-connect      | [2021-02-11 10:53:10,813] WARN Aborting multi-part upload with id '4d7da2e6e912fd41d4547ef57a056ab3' (io.confluent.connect.s3.storage.S3OutputStream)
  s3rver_1           | info: [S3rver] DELETE /test/topics/test/partition%3D0/test%2B0%2B0000000000.snappy.parquet?uploadId=4d7da2e6e912fd41d4547ef57a056ab3 204 - - 2.809 ms
  kafka-connect      | [2021-02-11 10:53:10,821] ERROR Multipart upload failed to complete for bucket 'test' key 'topics/test/partition=0/test+0+0000000000.snappy.parquet' (io.confluent.connect.s3.storage.S3OutputStream)
  kafka-connect      | [2021-02-11 10:53:10,821] ERROR Exception on topic partition test-0:  (io.confluent.connect.s3.S3SinkTask)
  kafka-connect      | org.apache.kafka.connect.errors.RetriableException: org.apache.kafka.connect.errors.DataException: Multipart upload failed to complete.
  kafka-connect      |    at io.confluent.connect.s3.TopicPartitionWriter.commitFiles(TopicPartitionWriter.java:506)
  kafka-connect      |    at io.confluent.connect.s3.TopicPartitionWriter.executeState(TopicPartitionWriter.java:230)
  kafka-connect      |    at io.confluent.connect.s3.TopicPartitionWriter.write(TopicPartitionWriter.java:188)
  kafka-connect      |    at io.confluent.connect.s3.S3SinkTask.put(S3SinkTask.java:190)
  </code></pre>
  <p>AS you can see I have my producer emitting an object, it seems it's writing a temporary object to s3 but then kafka-connect fails due to some &quot;multi part&quot; upload stuff. How could I make this work ? I don't really need the multi-part aspect for testing purposes ...</p>
  <p>Thanks a lot!</p>