KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter by rhauch · Pull Request #4815 · apache/kafka

rhauch · 2018-04-03T01:37:27Z

Summary of the problem

When the header.converter is not specified in the worker config or the connector config, a bug in the Plugins test causes it to never instantiate the HeaderConverter instance, even though there is a default value.

This is a problem as soon as the connector deals with headers, either in records created by a source connector or in messages on the Kafka topics consumed by a sink connector. As soon as that happens, a NPE occurs.

A workaround is to explicitly set the header.converter configuration property, but this was added in AK 1.1 and thus means that upgrading to AK 1.1 will not be backward compatible and will require this configuration change.

The Changes

The Plugins.newHeaderConverter methods were always returning null if the header.converter configuration value was not specified in the supplied connector or worker configuration. Thus, even though the header.converter property has a default, it was never being used.

The fix was to only check whether a header.converter property was specified when the connector configuration was being used, and if no such property exists in the connector configuration to return null. Then, when the worker configuration is being used, the method simply gets the header.converter value (or the default if no value was explicitly set).

Also, the ConnectorConfig had the same default value for the header.converter property as the WorkerConfig, but this resulted in very confusing log messages that implied the default header converter should be used even when the worker config specified the header.converter value. By removing the default, the log messages now make sense, and the Worker still properly instantiates the correct header converter.

Finally, updated comments and added log messages to make it more clear which converters are being used and how they are being converted.

Testing

Several new unit tests for Plugins.newHeaderConverter were added to check the various behavior. Additionally, the runtime JAR with these changes was built and inserted into an AK 1.1 installation, and a source connector was manually tested with 8 different combinations of settings for the header.converter configuration:

default value
worker configuration has header.converter explicitly set to the default
worker configuration has header.converter set to a custom HeaderConverter implementation in the same plugin
worker configuration has header.converter set to a custom HeaderConverter implementation in a different plugin
connector configuration has header.converter explicitly set to the default
connector configuration has header.converter set to a custom HeaderConverter implementation in the same plugin
connector configuration has header.converter set to a custom HeaderConverter implementation in a different plugin
worker configuration has header.converter explicitly set to the default, and the connector configuration has header.converter set to a custom HeaderConverter implementation in a different plugin

The worker created the correct HeaderConverter implementation with the correct configuration in all of these tests.

Finally, the default configuration was used with the aforementioned custom source connector that generated records with headers, and an S3 connector that consumes the records with headers (but didn't do anything with them). This test also passed.

Committer Checklist (excluded from commit message)

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

When the `header.converter` is not specified in the worker config or the connector config, a bug in the `Plugins` test causes it to never instantiate the `HeaderConverter` instance, even though there is a default value. This is a problem as soon as the connector deals with headers, either in records created by a source connector or in messages on the Kafka topics consumed by a sink connector. As soon as that happens, a NPE occurs. A workaround is to explicitly set the `header.converter` configuration property, but this was added in AK 1.1 and thus means that upgrading to AK 1.1 will not be backward compatible and will require this configuration change.

…r header converter The ConnectorConfig had the same default value for the `header.converter` property as the WorkerConfig, but this resulted in very confusing log messages that implied the default header converter should be used even when the worker config specified the `header.converter` value. By removing the default, the log messages now make sense, and the Worker still properly instantiates the correct header converter. Also updated comments and added log messages to make it more clear which converters are being used and how they are being converted.

rhauch · 2018-04-03T04:29:11Z

@ewencp Please review. This will need to be merged onto the 1.1 branch as well.

ewencp

@rhauch a few comments inline, I think the only "blockers" were log ordering things.

I think it's also worth calling out that the same check exists for newConverter in addition to the newHeaderConverter case. That happens to work currently because you are required to provide those configs. I wonder if (even if in follow up) we should make the same change for that path.

ewencp · 2018-04-03T04:38:53Z

                keyConverter = plugins.newConverter(config, WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.PLUGINS);
+                log.info("Set up the key converter {} for task {} using the worker config", keyConverter.getClass(), id);
+            } else {
+                log.info("Set up the key converter {} for task {} using the connector config", keyConverter.getClass(), id);


Nice, given we're even a little unsure exactly the right way for this to work, this logging should at least help us explain to users even if the behavior isn't entirely obvious.

ewencp · 2018-04-03T04:46:42Z

        HeaderConverter plugin = null;
        switch (classLoaderUsage) {
            case CURRENT_CLASSLOADER:
+                if (!config.originals().containsKey(classPropertyName)) {


might want some commentary here, or in the javadoc, about how this works. the CURRENT_CLASSLOADER name isn't really clear within this limited context to understand why it makes sense to skip in this case (connector overrides probably only make sense if it's explicit in the connector config) since the CURRENT_CLASSLOADER doesn't make it clear that it is the connector's classloader.

In fact, now I am wondering if we should rename the enums to be clearer. We can follow up on that separately, but I realize these branches aren't really very clear currently.

I'm concerned about doing that now. If you're okay with waiting, I can file a separate issue for later.

yeah, that's just about internal code being readable and understandable, not critical to current issue.

ewencp · 2018-04-03T04:54:29Z

        String configPrefix = classPropertyName + ".";
        Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
        plugin.configure(converterConfig, isKeyConverter);
+        log.debug("Configuring the {} converter with configuration:{}{}",


I think this should be above the previous line? Order of output could be confusing given current phrasing. Might also want to include isKeyConverter in the log.

Agree with moving. isKeyConverter is used in the first parameter in the log statement.

ewencp · 2018-04-03T04:54:48Z

        Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
        converterConfig.put(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName());
        plugin.configure(converterConfig);
+        log.debug("Configuring the header converter with configuration:{}{}", System.lineSeparator(), converterConfig);


same question as above about moving this above the call to configure

agree. Moving it before call to configure.

ewencp · 2018-04-03T05:02:31Z

    public static final String HEADER_CONVERTER_CLASS_DOC = WorkerConfig.HEADER_CONVERTER_CLASS_DOC;
    public static final String HEADER_CONVERTER_CLASS_DISPLAY = "Header converter class";
-    public static final String HEADER_CONVERTER_CLASS_DEFAULT = WorkerConfig.HEADER_CONVERTER_CLASS_DEFAULT;
+    public static final String HEADER_CONVERTER_CLASS_DEFAULT = null;


you might want a comment explaining the difference here, it'd be easy to interpret it as a mistaken deviation from WorkerConfig and "fix" it by restoring the inherited default.

rhauch · 2018-04-03T05:30:12Z

@ewencp I really wanted to make those changes to newConverter, but didn't due to timing. I will follow up with another issue.

rhauch · 2018-04-03T05:34:21Z

Logged the additional fixes/clarification as https://issues.apache.org/jira/browse/KAFKA-6740.

ewencp

LGTM, merging to trunk and 1.1

## Summary of the problem When the `header.converter` is not specified in the worker config or the connector config, a bug in the `Plugins` test causes it to never instantiate the `HeaderConverter` instance, even though there is a default value. This is a problem as soon as the connector deals with headers, either in records created by a source connector or in messages on the Kafka topics consumed by a sink connector. As soon as that happens, a NPE occurs. A workaround is to explicitly set the `header.converter` configuration property, but this was added in AK 1.1 and thus means that upgrading to AK 1.1 will not be backward compatible and will require this configuration change. ## The Changes The `Plugins.newHeaderConverter` methods were always returning null if the `header.converter` configuration value was not specified in the supplied connector or worker configuration. Thus, even though the `header.converter` property has a default, it was never being used. The fix was to only check whether a `header.converter` property was specified when the connector configuration was being used, and if no such property exists in the connector configuration to return null. Then, when the worker configuration is being used, the method simply gets the `header.converter` value (or the default if no value was explicitly set). Also, the ConnectorConfig had the same default value for the `header.converter` property as the WorkerConfig, but this resulted in very confusing log messages that implied the default header converter should be used even when the worker config specified the `header.converter` value. By removing the default, the log messages now make sense, and the Worker still properly instantiates the correct header converter. Finally, updated comments and added log messages to make it more clear which converters are being used and how they are being converted. ## Testing Several new unit tests for `Plugins.newHeaderConverter` were added to check the various behavior. Additionally, the runtime JAR with these changes was built and inserted into an AK 1.1 installation, and a source connector was manually tested with 8 different combinations of settings for the `header.converter` configuration: 1. default value 1. worker configuration has `header.converter` explicitly set to the default 1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin 1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin 1. connector configuration has `header.converter` explicitly set to the default 1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin 1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin 1. worker configuration has `header.converter` explicitly set to the default, and the connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin The worker created the correct `HeaderConverter` implementation with the correct configuration in all of these tests. Finally, the default configuration was used with the aforementioned custom source connector that generated records with headers, and an S3 connector that consumes the records with headers (but didn't do anything with them). This test also passed. Author: Randall Hauch <rhauch@gmail.com> Reviewers: Ewen Cheslack-Postava <ewen@confluent.io> Closes #4815 from rhauch/kafka-6728 (cherry picked from commit d9369de) Signed-off-by: Ewen Cheslack-Postava <me@ewencp.org>

## Summary of the problem When the `header.converter` is not specified in the worker config or the connector config, a bug in the `Plugins` test causes it to never instantiate the `HeaderConverter` instance, even though there is a default value. This is a problem as soon as the connector deals with headers, either in records created by a source connector or in messages on the Kafka topics consumed by a sink connector. As soon as that happens, a NPE occurs. A workaround is to explicitly set the `header.converter` configuration property, but this was added in AK 1.1 and thus means that upgrading to AK 1.1 will not be backward compatible and will require this configuration change. ## The Changes The `Plugins.newHeaderConverter` methods were always returning null if the `header.converter` configuration value was not specified in the supplied connector or worker configuration. Thus, even though the `header.converter` property has a default, it was never being used. The fix was to only check whether a `header.converter` property was specified when the connector configuration was being used, and if no such property exists in the connector configuration to return null. Then, when the worker configuration is being used, the method simply gets the `header.converter` value (or the default if no value was explicitly set). Also, the ConnectorConfig had the same default value for the `header.converter` property as the WorkerConfig, but this resulted in very confusing log messages that implied the default header converter should be used even when the worker config specified the `header.converter` value. By removing the default, the log messages now make sense, and the Worker still properly instantiates the correct header converter. Finally, updated comments and added log messages to make it more clear which converters are being used and how they are being converted. ## Testing Several new unit tests for `Plugins.newHeaderConverter` were added to check the various behavior. Additionally, the runtime JAR with these changes was built and inserted into an AK 1.1 installation, and a source connector was manually tested with 8 different combinations of settings for the `header.converter` configuration: 1. default value 1. worker configuration has `header.converter` explicitly set to the default 1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin 1. worker configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin 1. connector configuration has `header.converter` explicitly set to the default 1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in the same plugin 1. connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin 1. worker configuration has `header.converter` explicitly set to the default, and the connector configuration has `header.converter` set to a custom `HeaderConverter` implementation in a _different_ plugin The worker created the correct `HeaderConverter` implementation with the correct configuration in all of these tests. Finally, the default configuration was used with the aforementioned custom source connector that generated records with headers, and an S3 connector that consumes the records with headers (but didn't do anything with them). This test also passed. Author: Randall Hauch <rhauch@gmail.com> Reviewers: Ewen Cheslack-Postava <ewen@confluent.io> Closes apache#4815 from rhauch/kafka-6728 (cherry picked from commit d9369de) Signed-off-by: Ewen Cheslack-Postava <me@ewencp.org>

rhauch added 3 commits April 2, 2018 23:26

KAFKA-6728: Added more unit tests

0ffc79e

rhauch force-pushed the kafka-6728 branch from 5343371 to 864a52f Compare April 3, 2018 04:27

rhauch changed the base branch from 1.1 to trunk April 3, 2018 04:28

rhauch changed the title ~~[WIP] KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter~~ KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter Apr 3, 2018

ewencp reviewed Apr 3, 2018

View reviewed changes

KAFKA-6728: Addressed review comments

444d480

ewencp approved these changes Apr 3, 2018

View reviewed changes

ewencp closed this in d9369de Apr 3, 2018

kkonstantine added the connect label Oct 16, 2020

Conversation

rhauch commented Apr 3, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary of the problem

The Changes

Testing

Committer Checklist (excluded from commit message)

Uh oh!

rhauch commented Apr 3, 2018

Uh oh!

ewencp left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rhauch commented Apr 3, 2018

Uh oh!

rhauch commented Apr 3, 2018

Uh oh!

ewencp left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

rhauch commented Apr 3, 2018 •

edited

Loading