yashmayya commented on code in PR #14003:
URL: https://github.com/apache/kafka/pull/14003#discussion_r1261985349
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1608,6 +1612,32 @@ void modifySourceConnectorOffsets(String connName,
Connector connector, Map<Stri
}));
}
+ /**
+ * "Normalize" source connector offsets by serializing and deserializing
them using the internal {@link JsonConverter}.
+ * This is done in order to prevent type mismatches between the offsets
passed to {@link SourceConnector#alterOffsets(Map, Map)}
+ * and the offsets that connectors and tasks retrieve via an instance of
{@link OffsetStorageReader}.
+ * <p>
+ * Visible for testing.
+ *
+ * @param originalOffsets the offsets that are to be normalized
+ * @return the normalized offsets
+ */
+ @SuppressWarnings("unchecked")
+ Map<Map<String, ?>, Map<String, ?>>
normalizeSourceConnectorOffsets(Map<Map<String, ?>, Map<String, ?>>
originalOffsets) {
+ Map<Map<String, ?>, Map<String, ?>> normalizedOffsets = new
HashMap<>();
+ for (Map.Entry<Map<String, ?>, Map<String, ?>> entry :
originalOffsets.entrySet()) {
+ OffsetUtils.validateFormat(entry.getKey());
+ OffsetUtils.validateFormat(entry.getValue());
+ byte[] serializedKey = internalKeyConverter.fromConnectData("",
null, entry.getKey());
Review Comment:
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));
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
