[KAFKA-15182] Normalize offsets before invoking SourceConnector::alterOffsets

• Type: Improvement
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.6.0
• Fix Version/s: 3.6.0
• Component/s: connect
• Assignee: Yash Mayya
• Reporter: Yash Mayya
• Created: 12/Jul/23 17:36
• Updated: 14/Jul/23 16:57
• Resolved: 14/Jul/23 16:14

Description

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.

Issue Links

• links to: GitHub Pull Request #14003