feat: allow null and empty headers configuration option

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

@@ -178,6 +178,8 @@ public class S3SinkConnectorConfig extends StorageSinkConnectorConfig {
 
   public static final String BEHAVIOR_ON_NULL_VALUES_CONFIG = "behavior.on.null.values";
   public static final String BEHAVIOR_ON_NULL_VALUES_DEFAULT = OutputWriteBehavior.FAIL.toString();
+  public static final String ALLOW_NULL_AND_EMPTY_HEADERS_CONFIG = "allow.null.and.empty.headers";
+  public static final Boolean ALLOW_NULL_AND_EMPTY_HEADERS_DEFAULT = false;
 
   public static final String REPORT_NULL_RECORDS_TO_DLQ = "report.null.values.to.dlq";
   public static final boolean REPORT_NULL_RECORDS_TO_DLQ_DEFAULT = true;
@@ -724,6 +726,18 @@ public static ConfigDef newConfigDef() {
           "Behavior for null-valued records"
       );
 
+      configDef.define(
+              ALLOW_NULL_AND_EMPTY_HEADERS_CONFIG,
+              Type.BOOLEAN,
+              ALLOW_NULL_AND_EMPTY_HEADERS_DEFAULT,
+              Importance.LOW,
+              "Whether to allow null and empty headers when writing headers is enabled.",
+              group,
+              ++orderInGroup,
+              Width.SHORT,
+              "Whether to allow null and empty headers when writing headers is enabled."
+      );
+
       configDef.define(
           REPORT_NULL_RECORDS_TO_DLQ,
           Type.BOOLEAN,
@@ -1424,6 +1438,10 @@ public boolean shouldRotateOnPartitionChange() {
     return getBoolean(ROTATE_FILE_ON_PARTITION_CHANGE);
   }
 
+  public Boolean allowNullAndEmptyHeaders() {
+    return getBoolean(ALLOW_NULL_AND_EMPTY_HEADERS_CONFIG);
+  }
+
   public enum IgnoreOrFailBehavior {
     IGNORE,
     FAIL;

kafka-connect-s3/src/main/java/io/confluent/connect/s3/format/KeyValueHeaderRecordWriterProvider.java

@@ -103,6 +103,7 @@ public void write(SinkRecord sinkRecord) {
 
         // headerWriter != null means writing headers is turned on
         if (headerWriter.isPresent()
+            && !conf.allowNullAndEmptyHeaders()
             && (sinkRecord.headers() == null || sinkRecord.headers().isEmpty())) {
           throw new DataException(
               String.format("Headers cannot be null for SinkRecord: %s",

kafka-connect-s3/src/test/java/io/confluent/connect/s3/S3SinkTaskTest.java

@@ -145,6 +145,33 @@ public void testWriteNullRecords() throws Exception {
     Mockito.verify(reporter, times(2)).report(any(), any(DataException.class));
   }
 
+  @Test
+  public void testAllowNullAndEmptyHeaders() throws Exception {
+    setUp();
+    replayAll();
+    task = new S3SinkTask();
+    task.initialize(context);
+    properties.put(S3SinkConnectorConfig.ALLOW_NULL_AND_EMPTY_HEADERS_CONFIG, "true");
+    task.start(properties);
+    verifyAll();
+
+    List<SinkRecord> sinkRecords = createRecords(7);
+    SinkRecord recordWithNullHeaders = new SinkRecord(TOPIC, PARTITION, null, null, null, null, 1, null, null, Collections.emptyList());
+    SinkRecord recordWithEmptyHeaders = new SinkRecord(TOPIC, PARTITION, null, null, null, null, 1, null, null, null);
+
+    sinkRecords.add(recordWithNullHeaders);
+    sinkRecords.add(recordWithEmptyHeaders);
+
+    task.put(sinkRecords);
+
+    task.close(context.assignment());
+    task.stop();
+
+    long[] validOffsets = {0, 3, 6};
+
+    verify(sinkRecords, validOffsets);
+  }
+
   @Test
   public void testWriteRecordWithPrimitives() throws Exception {
     setUp();