@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());
            byte[] serializedValue = internalValueConverter.fromConnectData("", null, entry.getValue());
            Object deserializedKey = internalKeyConverter.toConnectData("", serializedKey).value();
            Object deserializedValue = internalValueConverter.toConnectData("", serializedValue).value();
            normalizedOffsets.put((Map<String, ?>) deserializedKey, (Map<String, ?>) deserializedValue);
        }

        return normalizedOffsets;
    }
"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
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java