Blocker Description
When a connector request is submitted that overrides a client configuration that is meant to contain the name of a class (such as sasl.login.callback.handler.class), a 500 response is generated and the following stack trace can be found in the logs for Connect:
[2019-05-22 14:51:36,123] ERROR Uncaught exception in REST call to /connectors (org.apache.kafka.connect.runtime.rest.errors.ConnectExceptionMapper:61)
java.lang.ClassCastException: java.lang.String cannot be cast to java.lang.Class
at org.apache.kafka.common.config.ConfigDef.convertToString(ConfigDef.java:774)
at org.apache.kafka.connect.runtime.AbstractHerder.convertConfigValue(AbstractHerder.java:491)
at org.apache.kafka.connect.runtime.AbstractHerder.validateClientOverrides(AbstractHerder.java:426)
at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:342)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$6.call(DistributedHerder.java:565)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder$6.call(DistributedHerder.java:562)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.tick(DistributedHerder.java:292)
at org.apache.kafka.connect.runtime.distributed.DistributedHerder.run(DistributedHerder.java:241)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
This appears to be limited only to client configs that are meant to be classes or lists due to the fact that ConfigDef.convertToString(...) assumes its first argument is an instance of Class<?> when its second argument is ConfigDef.Type.CLASS and then casts accordingly, and acts similarly for lists. If the second argument is anything else, toString() is invoked on it without any casting, avoiding any problems.
The cause of this is due to the fact that the newly-introduced ConnectorClientConfigOverridePolicy interface returns a list of ConfigValue instances for its validation. The value() for each of these can be any type, although with the default implementations available (All, None, Principal) if one is returned at all it's just the same type of what was passed in for that particular config. In the case of the AbstractHerder.validateClientOverrides(...) method, the raw strings for the client configs are used. However, the AbstractHerder.convertConfigValue(...) is then called for those raw strings but with the ConfigDef.Type of the config based on the relevant client ConfigDef (i.e., ProducerConfig.configDef(), ConsumerConfig.configDef(), or AdminClientConfig.configDef()). This in turn can and will result in ConfigDef.convertToString(someClassNameAsAString, ConfigDef.Type.CLASS) being invoked.
Although this isn't technically a comprehensive fix, a quick option would be to invoke ConfigDef.parse(...) using the relevant client ConfigDef before passing overrides to the policy. Technically, this would still lead to problems if the policy decided to return just the name of a class for a config that of type class instead, so we may want to investigate other options as well.
Attachments
Issue Links
- links to