KAFKA-8265: Initial implementation for ConnectorClientConfigPolicy to enable overrides (KIP-458)

[mageshn]

Implementation to enable policy for Connector Client config overrides. This is implemented per the KIP-458.

Committer Checklist (excluded from commit message)

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

[rhauch]

Thanks, @mageshn! I have a number of questions and suggestions below.

[rhauch]

By not including these prefixes, might some of these collide where the values are distinct? Also, we need to be careful that we don't pass non-producer properties to the ProducerConfig, or else it will log them as a warning. Won't excluding the prefixes here make it more difficult to prevent those warnings?

[mageshn]

These are the configs that are passed to policy implementation. So, I'm not sure I completely understand your concern here.

[rhauch]

I guess what seems strange is that we're removing the prefixes for the client overrides, but otherwise this is the raw connector configuration, at which point some of the overrides might conflict with some of the connector-specific configs. Also, some of the producer overrides might conflict with some of the admin overrides. Isn't the point of having the client override prefixes is to make sure we have no conflicts, and aren't we losing that here?

[mageshn]

But we are returning just the client configs here not the complete connector configs. So, there are no conflicts here.

[rhauch]

Seems like there's a fair amount of boilerplate here, amongst all of the policy unit tests. How about a base class that defines an validateWithPolicy(clientConfig) (or something similar), so that it's more clear what each test is checking.

[rhauch]

Nice work. A few more comments, and I resolved most of the questions/conversations.

[rhauch]

Missing a period between sentences, and have an extra > character.

Also seems like we should mention that custom implementations are allowed.

[rhauch]

That was not clear from the KIP or the interface docs, because the request provides a map of values but it's always a singleton. If I were implementing a policy, I would not have expected one property at a time. Plus, this provides no way for the implementation to actually check any combination of values, should that ever be required (we don't use that in the built-in implementations).

I'm not sure what can be done at this point, given the use of PVE to pass the message. It would have been much better to pass in a component with which the implementation could have recorded errors for one or more configuration properties.

[rhauch]

Nice job, @mageshn! I like this direction much more, and it's much cleaner to implement a policy, and this also gives us the opportunity to implement a more complex policy should it be needed.

A few suggestions below.

[rhauch]

We need unit tests for this.

[rhauch]

One minor correction, but otherwise looks good.

[rhauch]

I rebased on the latest trunk to fix a conflict in DistributedHerderTest.java.

[rhauch]

Thanks, @mageshn! This is a nice feature and nice implementation of KIP-458 for the next AK release.

LGTM, pending a green build.

[ijuma]

One job passed and the other looked like it was going to pass and then it ran into what looks like a gradle bug:

08:29:42 ERROR: Failed to write output for test null.Gradle Test Executor 19
08:29:42 java.lang.NullPointerException: Cannot invoke method write() on null object
08:29:42 at org.codehaus.groovy.runtime.NullObject.invokeMethod(NullObject.java:91)
08:29:42 at org.codehaus.groovy.runtime.callsite.PogoMetaClassSite.call(PogoMetaClassSite.java:47)
08:29:42 at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:47)
08:29:42 at org.codehaus.groovy.runtime.callsite.NullCallSite.call(NullCallSite.java:34)
08:29:42 at org.codehaus.groovy.runtime.callsite.CallSiteArray.defaultCall(CallSiteArray.java:47)
08:29:42 at java_io_FileOutputStream$write.call(Unknown Source)

Will go ahead and merge since @rhauch approved.

[ijuma]

Sorry, I saw this late @rhauch and @mageshn. We should not do this as these types are mutable and anyone could change the shared definition.

[mageshn]

That's a good point @ijuma. What do you think about returning a copy like
public static ConfigDef configDef() {return new ConfigDef(CONFIG); }