Repro 2

checkstyle/suppressions.xml

@@ -21,4 +21,9 @@
       files="(TopicPartitionWriter).java"
     />
 
+    <suppress
+            checks="NCSS"
+            files="(S3SinkConnectorConfig).java"
+    />
+
 </suppressions>

kafka-connect-s3/src/main/java/io/confluent/connect/s3/S3SinkConnectorConfig.java

@@ -49,6 +49,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
+import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Function;
@@ -200,7 +201,7 @@ public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
   public static final Class<? extends Format> HEADERS_FORMAT_CLASS_DEFAULT = AvroFormat.class;
 
   /**
-   * Elastic buffer to save memory. {@link io.confluent.connect.s3.storage.S3OutputStream#buffer}
+   * Elastic buffer to save memory. {@link io.confluent.connect.s3.storage.S3OutputStream}
    */
 
   public static final String ELASTIC_BUFFER_ENABLE = "s3.elastic.buffer.enable";
@@ -223,6 +224,11 @@ public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
       + "to prefix or suffix in the s3 path after the topic name."
       + " None will not append the schema name in the s3 path.";
 
+
+  private static final String SHOULD_SLEEP_ON_COMMIT = "sleep.on.commit";
+  private static final String SLEEP_INTERVAL = "sleep.interval";
+  private static final String SLEEP_PROBABILITY = "sleep.probability";
+
   private static final GenericRecommender SCHEMA_PARTITION_AFFIX_TYPE_RECOMMENDER =
       new GenericRecommender();
 
@@ -831,6 +837,8 @@ public static ConfigDef newConfigDef() {
           "Elastic buffer initial capacity"
       );
 
+      addZombieConfigs(configDef);
+
     }
     return configDef;
   }
@@ -869,6 +877,46 @@ private void validateTimezone() {
     }
   }
 
+  private static void addZombieConfigs(ConfigDef config) {
+    String group = "zombie";
+    int orderInGroup = 0;
+    config.define(
+        SLEEP_INTERVAL,
+        Type.LONG,
+        -1,
+        Importance.HIGH,
+        "Sleep interval millis",
+        group,
+        ++orderInGroup,
+        Width.LONG,
+        "Sleep interval millis"
+    );
+
+    config.define(
+        SHOULD_SLEEP_ON_COMMIT,
+        Type.BOOLEAN,
+        false,
+        Importance.HIGH,
+        "Should sleep on commit",
+        group,
+        ++orderInGroup,
+        Width.LONG,
+        "Should sleep on commit"
+    );
+
+    config.define(
+        SLEEP_PROBABILITY,
+        Type.INT,
+        2,
+        Importance.HIGH,
+        "Probability index",
+        group,
+        ++orderInGroup,
+        Width.LONG,
+        "Probability index"
+    );
+  }
+
   private void addToGlobal(AbstractConfig config) {
     allConfigs.add(config);
     addConfig(config.values(), (ComposableConfig) config);
@@ -1013,6 +1061,16 @@ public boolean getElasticBufferEnable() {
     return getBoolean(ELASTIC_BUFFER_ENABLE);
   }
 
+  private Random random = new Random();
+
+  public boolean getShouldSleep() {
+    return getBoolean(SHOULD_SLEEP_ON_COMMIT) && random.nextInt() % getInt(SLEEP_PROBABILITY) == 0;
+  }
+
+  public long sleepDuartion() {
+    return getLong(SLEEP_INTERVAL);
+  }
+
   public int getElasticBufferInitCap() {
     return getInt(ELASTIC_BUFFER_INIT_CAPACITY);
   }