Skip to content

KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter #4815

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
rhauch wants to merge 4 commits into from
Closed

KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter #4815

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
8caef16
KAFKA-6728: Corrected the worker’s instantiation of the HeaderConverter
rhauch Apr 3, 2018
0ffc79e
KAFKA-6728: Added more unit tests
rhauch Apr 3, 2018
864a52f
KAFKA-6728: Added log messages and removed ConnectorConfig default fo…
rhauch Apr 3, 2018
444d480
KAFKA-6728: Addressed review comments
rhauch Apr 3, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,9 @@ public class ConnectorConfig extends AbstractConfig {
public static final String HEADER_CONVERTER_CLASS_CONFIG = WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG;
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;
// The Connector config should not have a default for the header converter, since the absence of a config property means that
// the worker config settings should be used. Thus, we set the default to null here.
public static final String HEADER_CONVERTER_CLASS_DEFAULT = null;

public static final String TASKS_MAX_CONFIG = "tasks.max";
private static final String TASKS_MAX_DOC = "Maximum number of tasks to use for this connector.";
Expand Down
9 changes: 9 additions & 0 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -397,12 +397,21 @@ public boolean startTask(
);
if (keyConverter == null) {
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);

@ewencp ewencp Apr 3, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
if (valueConverter == null) {
valueConverter = plugins.newConverter(config, WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.PLUGINS);
log.info("Set up the value converter {} for task {} using the worker config", valueConverter.getClass(), id);
} else {
log.info("Set up the value converter {} for task {} using the connector config", valueConverter.getClass(), id);
}
if (headerConverter == null) {
headerConverter = plugins.newHeaderConverter(config, WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG, ClassLoaderUsage.PLUGINS);
log.info("Set up the header converter {} for task {} using the worker config", headerConverter.getClass(), id);
} else {
log.info("Set up the header converter {} for task {} using the connector config", headerConverter.getClass(), id);
}

workerTask = buildWorkerTask(connConfig, id, task, statusListener, initialState, keyConverter, valueConverter, headerConverter, connectorLoader);
Expand Down
14 changes: 9 additions & 5 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -234,6 +234,8 @@ public Converter newConverter(AbstractConfig config, String classPropertyName, C
// Configure the Converter using only the old configuration mechanism ...
String configPrefix = classPropertyName + ".";
Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
log.debug("Configuring the {} converter with configuration:{}{}",
isKeyConverter ? "key" : "value", System.lineSeparator(), converterConfig);
plugin.configure(converterConfig, isKeyConverter);
return plugin;
}
Expand All @@ -249,20 +251,21 @@ public Converter newConverter(AbstractConfig config, String classPropertyName, C
* @throws ConnectException if the {@link HeaderConverter} implementation class could not be found
*/
public HeaderConverter newHeaderConverter(AbstractConfig config, String classPropertyName, ClassLoaderUsage classLoaderUsage) {
if (!config.originals().containsKey(classPropertyName)) {
// This configuration does not define the header converter via the specified property name
return null;
}
HeaderConverter plugin = null;
switch (classLoaderUsage) {
case CURRENT_CLASSLOADER:
if (!config.originals().containsKey(classPropertyName)) {

@ewencp ewencp Apr 3, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@ewencp ewencp Apr 3, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@rhauch rhauch Apr 3, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@ewencp ewencp Apr 3, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

// This connector configuration does not define the header converter via the specified property name
return null;
}
// Attempt to load first with the current classloader, and plugins as a fallback.
// Note: we can't use config.getConfiguredInstance because we have to remove the property prefixes
// before calling config(...)
plugin = getInstance(config, classPropertyName, HeaderConverter.class);
break;
case PLUGINS:
// Attempt to load with the plugin class loader, which uses the current classloader as a fallback
// Attempt to load with the plugin class loader, which uses the current classloader as a fallback.
// Note that there will always be at least a default header converter for the worker
String converterClassOrAlias = config.getClass(classPropertyName).getName();
Class<? extends HeaderConverter> klass;
try {
Expand All @@ -288,6 +291,7 @@ public HeaderConverter newHeaderConverter(AbstractConfig config, String classPro
String configPrefix = classPropertyName + ".";
Map<String, Object> converterConfig = config.originalsWithPrefix(configPrefix);
converterConfig.put(ConverterConfig.TYPE_CONFIG, ConverterType.HEADER.getName());
log.debug("Configuring the header converter with configuration:{}{}", System.lineSeparator(), converterConfig);
plugin.configure(converterConfig);
return plugin;
}
Expand Down
76 changes: 59 additions & 17 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/isolation/PluginsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.kafka.connect.storage.ConverterConfig;
import org.apache.kafka.connect.storage.ConverterType;
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.storage.SimpleHeaderConverter;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
Expand All @@ -39,18 +40,31 @@

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;

public class PluginsTest {

private static Map<String, String> props;
private static Map<String, String> pluginProps;
private static Plugins plugins;
private Map<String, String> props;
private AbstractConfig config;
private TestConverter converter;
private TestHeaderConverter headerConverter;

@BeforeClass
public static void beforeAll() {
props = new HashMap<>();
pluginProps = new HashMap<>();

// Set up the plugins to have no additional plugin directories.
// This won't allow us to test classpath isolation, but it will allow us to test some of the utility methods.
pluginProps.put(WorkerConfig.PLUGIN_PATH_CONFIG, "");
plugins = new Plugins(pluginProps);
}

@Before
public void setup() {
props = new HashMap<>(pluginProps);
props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, TestConverter.class.getName());
props.put(WorkerConfig.VALUE_CONVERTER_CLASS_CONFIG, TestConverter.class.getName());
props.put("key.converter." + JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "true");
Expand All @@ -66,14 +80,10 @@ public static void beforeAll() {
props.put(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG, TestHeaderConverter.class.getName());
props.put("header.converter.extra.config", "baz");

// Set up the plugins to have no additional plugin directories.
// This won't allow us to test classpath isolation, but it will allow us to test some of the utility methods.
props.put(WorkerConfig.PLUGIN_PATH_CONFIG, "");
plugins = new Plugins(props);
createConfig();
}

@Before
public void setup() {
protected void createConfig() {
this.config = new TestableWorkerConfig(props);
}

Expand Down Expand Up @@ -104,23 +114,55 @@ public void shouldInstantiateAndConfigureInternalConverters() {
}

@Test
public void shouldInstantiateAndConfigureHeaderConverter() {
instantiateAndConfigureHeaderConverter(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG);
public void shouldInstantiateAndConfigureExplicitlySetHeaderConverterWithCurrentClassLoader() {
assertNotNull(props.get(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG));
HeaderConverter headerConverter = plugins.newHeaderConverter(config,
WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG,
ClassLoaderUsage.CURRENT_CLASSLOADER);
assertNotNull(headerConverter);
assertTrue(headerConverter instanceof TestHeaderConverter);
this.headerConverter = (TestHeaderConverter) headerConverter;

// Validate extra configs got passed through to overridden converters
assertConverterType(ConverterType.HEADER, this.headerConverter.configs);
assertEquals("baz", this.headerConverter.configs.get("extra.config"));

headerConverter = plugins.newHeaderConverter(config,
WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG,
ClassLoaderUsage.PLUGINS);
assertNotNull(headerConverter);
assertTrue(headerConverter instanceof TestHeaderConverter);
this.headerConverter = (TestHeaderConverter) headerConverter;

// Validate extra configs got passed through to overridden converters
assertConverterType(ConverterType.HEADER, headerConverter.configs);
assertEquals("baz", headerConverter.configs.get("extra.config"));
assertConverterType(ConverterType.HEADER, this.headerConverter.configs);
assertEquals("baz", this.headerConverter.configs.get("extra.config"));
}

@Test
public void shouldInstantiateAndConfigureDefaultHeaderConverter() {
props.remove(WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG);
createConfig();

// Because it's not explicitly set on the supplied configuration, the logic to use the current classloader for the connector
// will exit immediately, and so this method always returns null
HeaderConverter headerConverter = plugins.newHeaderConverter(config,
WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG,
ClassLoaderUsage.CURRENT_CLASSLOADER);
assertNull(headerConverter);
// But we should always find it (or the worker's default) when using the plugins classloader ...
headerConverter = plugins.newHeaderConverter(config,
WorkerConfig.HEADER_CONVERTER_CLASS_CONFIG,
ClassLoaderUsage.PLUGINS);
assertNotNull(headerConverter);
assertTrue(headerConverter instanceof SimpleHeaderConverter);
}

protected void instantiateAndConfigureConverter(String configPropName, ClassLoaderUsage classLoaderUsage) {
converter = (TestConverter) plugins.newConverter(config, configPropName, classLoaderUsage);
assertNotNull(converter);
}

protected void instantiateAndConfigureHeaderConverter(String configPropName) {
headerConverter = (TestHeaderConverter) plugins.newHeaderConverter(config, configPropName, ClassLoaderUsage.CURRENT_CLASSLOADER);
assertNotNull(headerConverter);
}

protected void assertConverterType(ConverterType type, Map<String, ?> props) {
assertEquals(type.getName(), props.get(ConverterConfig.TYPE_CONFIG));
}
Expand Down