feat: allow null and empty headers configuration option

[Nick Sia (itsnicksia)]

Problem

S3 Sink Connector will throw an error when processing a message with missing/empty headers when write headers is enabled.

In certain cases, users may wish to process topics that have a mix of messages with and without headers, and to save the headers where they exist.

For example:
#602

Solution

Provide a configuration option to allowmessages with missing headers, even when write headers is enabled.

The expected behaviour is that the connector will simply not write the header if it is empty or missing.

The default behaviour state of the setting will be false, to maintain current behaviour of throwing an exception.

Does this solution apply anywhere else?

• yes
• no

Test Strategy

Manual testing on developer device.

Testing done:

• Unit tests
• Integration tests
• System tests
• Manual tests

Release Plan

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
✅ itsnicksia
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}

[Copilot]

Pull Request Overview

This PR adds a new configuration option to allow the S3 Sink Connector to process messages with null or empty headers when header writing is enabled. Previously, the connector would throw an error when encountering messages without headers even if header writing was configured.

• Adds allow.null.and.empty.headers configuration option with default value false
• Modifies header validation logic to skip the error when the new option is enabled
• Includes unit test to verify the new behavior works correctly

Reviewed Changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
S3SinkConnectorConfig.java	Adds new configuration option definition and accessor method
KeyValueHeaderRecordWriterProvider.java	Updates header validation logic to check the new configuration
S3SinkTaskTest.java	Adds unit test for the new null/empty headers functionality

[Nick Sia (nicksia-vgw)]

Subhash Iyer (@subhashiyer9) Sorry to bother - any chance I could get some feedback on this change please?