KAFKA-15182: Normalize source connector offsets before invoking SourceConnector::alterOffsets

[yashmayya]

• From https://issues.apache.org/jira/browse/KAFKA-15182:

See discussion here

TLDR: When users attempt to externally modify source connector offsets via the PATCH /offsets endpoint (introduced in KIP-875), type mismatches can occur between offsets passed to SourceConnector::alterOffsets and the offsets that are retrieved by connectors / tasks via an instance of OffsetStorageReader after the offsets have been modified. In order to prevent this type mismatch that could lead to subtle bugs in connectors, we could serialize + deserialize the offsets using the worker's internal JSON converter before invoking SourceConnector::alterOffsets.

• I've also added a small unit test, verified that the existing offsets API related integration tests are passing, and tested this patch out manually with the FileStreamSourceConnector.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[yashmayya]

This should be safe to do because the OffsetStorageReaderImpl also serializes the connector / task specified source partition before retrieving its corresponding source offset. The difference here is that there is an extra ser / deser hop although that shouldn't cause issues. So, for instance:

Map<String, Object> p1 = Collections.singletonMap("partition_key", 10);
Map<String, Object> p2 = Collections.singletonMap("partition_key", 10L);

ByteBuffer serializedP1 = ByteBuffer.wrap(converter.fromConnectData("", null, p1));
ByteBuffer serializedP2 = ByteBuffer.wrap(converter.fromConnectData("", null, p2));

assertTrue(serializedP1.equals(serializedP2));

[C0urante]

Thanks Yash! Looks great, some minor thoughts and then this should be good to go.

[C0urante]

LGTM, thanks Yash!