Error with retrieving key of SinkRecords

[subkanthi]

Currently if the key is a byte array and if its not UTF-8 encoded, the key is not retrieved.

    private String getKeyString(SinkRecord record) {
        String keyString = null;
        final Object recordKey = record.key();
        if (recordKey != null) {
            if (recordKey instanceof byte[] && ByteArrayUtils.isUTF8Encoded((byte[]) recordKey)) {
                keyString = new String((byte[]) recordKey, StandardCharsets.UTF_8);
            } else if (recordKey instanceof String) {
                keyString = (String) recordKey;
            }
        }
        return keyString;
    }

┆Issue is synchronized with this Jira Story by Unito

[sync-by-unito]

➤ Automation for Jira commented:

The link to the corresponding Jira issue is https://ably.atlassian.net/browse/SDK-3579

[jaley]

@subkanthi - that's a private method in ConfigValueEvaluator, and if it returns null, the calling code checks to see whether the key is referenced in the mapping string. If it is, then that's an error, because we need to be able to interpret the key as a string in order to use it in the channel mapping. There's a config to optionally skip records with keys we can't use.

Arguably, the code should check whether we're using #{key} in the mapping before extracting a key, as it's a waste to check every record to see if it's UTF-8 if it's not being used anyway. We could potentially also support other string encodings, like UTF-16.

What were you finding is broken by the above code? What's the expected behaviour?

[subkanthi]

@jaley

recordKey would not meet the if and else if conditions, its a byte[] array but not UTF8 encoded and so the keyString will be null.

            if (recordKey instanceof byte[] && ByteArrayUtils.isUTF8Encoded((byte[]) recordKey)) {
                keyString = new String((byte[]) recordKey, StandardCharsets.UTF_8);
            } else if (recordKey instanceof String) {
                keyString = (String) recordKey;
            }

@subkanthi - that's a private method in ConfigValueEvaluator, and if it returns null, the calling code checks to see whether the key is referenced in the mapping string. If it is, then that's an error, because we need to be able to interpret the key as a string in order to use it in the channel mapping. There's a config to optionally skip records with keys we can't use.

Arguably, the code should check whether we're using #{key} in the mapping before extracting a key, as it's a waste to check every record to see if it's UTF-8 if it's not being used anyway. We could potentially also support other string encodings, like
UTF-16.

What were you finding is broken by the above code? What's the expected behaviour?