KAFKA-6685: Added Exception to distinguish message Key from Value during deserializing.

[jadireddi]

https://issues.apache.org/jira/browse/KAFKA-6685

Added Exception message in WorkerSinkTask.convertMessages to distinguish message Key from Value during deserialization to Kafka connect format.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

[jadireddi]

retest this please

[jadireddi]

Hi @rhauch,
could you please help in assigning Reviewers to review this ticket. I missed this during Pull request creation.

[rhauch]

@jadireddi thanks for creating this PR and jumping in to add the better exception logic. I do have a suggestion for making this more readable and concise.

[rhauch]

Perhaps the following would be a bit more straightforward without the boolean flag while providing more meaningful content in the error message?

        final SchemaAndValue keyAndSchema;
        final SchemaAndValue valueAndSchema;
        try {
            keyAndSchema = keyConverter.toConnectData(msg.topic(), msg.key());
        } catch (Throwable e) {
            String str = String.format("Error converting message key in topic '%s' partition %d at offset %d and timestamp %d",
                                       msg.topic(), msg.partition(), msg.offset(), msg.timestamp());
            throw new ConnectException(str, e);
        }
        try {
            valueAndSchema = valueConverter.toConnectData(msg.topic(), msg.value());
        } catch (Throwable e) {
            String str = String.format("Error converting message value in topic '%s' partition %d at offset %d and timestamp %d",
                                       msg.topic(), msg.partition(), msg.offset(), msg.timestamp());
            throw new ConnectException(str, e);
        }

Also, note that we don't want to log the exception here since exceptions will ultimately be caught and logged in WorkerTask.doRun(). Logging it here is unnecessary.

And actually, since there's a fair amount of duplicate logic, a better way of reducing that logic is to extract it into a method:

        SchemaAndValue keyAndSchema = toConnectData(keyConverter, "key", msg, msg.key());
        SchemaAndValue valueAndSchema = toConnectData(valueConverter, "value", msg, msg.value());

with the method:

private SchemaAndValue toConnectData(Converter converter, String converterName, ConsumerRecord<byte[], byte[]> msg, byte[] data) {
    try {
        return converter.toConnectData(msg.topic(), data);
    } catch (Throwable e) {
        String str = String.format("Error converting message %s in topic '%s' partition %d at offset %d and timestamp %d",
                                   converterName, msg.topic(), msg.partition(), msg.offset(), msg.timestamp());
        throw new ConnectException(str, e);
    }
}

This is even more concise and readable. WDYT?

[jadireddi]

Thank you @rhauch for your time to review and provide valuable suggestions. Code looks really concise and readable. Making changes.

[jadireddi]

Hi @rhauch , Incorporated review comments. Kindly review them.

[jadireddi]

@rhauch, Can you please review changes and let me know if anything needed.

[jadireddi]

retest this please

[rhauch]

@jadireddi one more suggestion, but otherwise looks good.

[rhauch]

How about just declaring and initializing each variable in one statement?

SchemaAndValue keyAndSchema = toConnectData(keyConverter, "key", msg, msg.key());
SchemaAndValue valueAndSchema = toConnectData(valueConverter, "value", msg, msg.value());

Also, in this case I'm not sure final adds much if any value.

[jadireddi]

@rhauch, Fixed review comments. It's true final doesn't add much value in this case..Removed it. Thank you for the review comments..

[rhauch]

LGTM. @ewencp, would you mind reviewing and merging if all looks good?

Thanks!

[ewencp]

LGTM, thanks for the contribution to debuggability!