KIP-476: Add new getAdmin method to KafkaClientSupplier

[big-andy-coates]

KIP-476 saw the introduction of a new Admin interface to replace the use of the AdminClient abstract class. A key driver for this was to allow the use of dynamic proxies for the admin client supplied to Kafka streams via the KafkaClientSupplier.

Unfortunately, it would be a binary incompatible change to change KafkClientSupplier.getAdminClient to return the Admin interface, rather than AdminClient. See #7157 which reverted the change.

To avoid this, I've added a new method getAdmin and deprecated the old getAdminClient.

The new getAdmin method has a default implementation that simply calls the old getAdminClient method. Both DefaultKafkaClientSupplier and MockClientSupplier implement the new method. All KS code has been switched to call the new method.

This has the benefit of being binary compatible, but does leave users still needing to implement the, now deprecated, getAdminClient method, even though it's not called by KS code.

Thoughts?

Committer Checklist (excluded from commit message)

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

[mjsax]

Overall LGTM.

Note, that this is a public API change and should be included in the KIP. Hence, please update the KIP accordingly and send an "FYI" e-mail to the VOTE thread to point out this minor change (don't think we need to re-vote; it's just to give people a notification in case anyone want to change their vote)

[mjsax]

The comment is a little "miss leading" because one still needs to implement this method... Should be add a default implementation that throws an exception?

[big-andy-coates]

I'll switch this to have a default impl that throws:

throw new UnsupportedOperationException("Direct use of this method is deprecated. The method will be removed in a future release.");

This means anyone new implementing the interface won't need to implement this method, and calling it will throw, but for any existing implementations they can continue to implement and call as before, though it will generate compiler warnings as its deprecated.

The only downside of this is that now both the old getAdminClient and new getAdmin have default implementations that throw. I guess this is OK, as when we remove the old method we can remove the default impl from the new method.

[mjsax]

That makes sense.

[ijuma]

I think the main downside is that type safety was lost. The compiler doesn't help you if you forget to implement the method.

[big-andy-coates]

I agree, but...

Most people use the inbuilt suppliers, so aren't affected.

Existing users that have implemented their own versions will work as-is, though they'll get a deprecated warning, which they can choose to clean up by changing getAdminClient to getAdmin in their impl, i.e. minor change.

New users might not initially implement either method, but will quickly run into runtime exceptions and rectify the situation.

Plus, this state is only maintained until the next major version bump, (which may be a long time of course), where we can then clean up the old and remove the default from the new.

Thoughts?

[ijuma]

Assuming this interface is not implemented often, that seems OK.

[big-andy-coates]

retest this please

[mjsax]

One nit to make the error message more readable.

Overall LGTM.

Call for second review @guozhangwang @bbejeck @vvcephei @ableegoldman @cadonna @abbccdda

[mjsax]

nit: getAdmin method -> getAdmin() method ?

[ijuma]

This is an incompatible change, why are we doing it in a minor release?

[ijuma]

I see, the assumption is that existing implementations will have overriden it.

[big-andy-coates]

Yeah, it's not incompatible with our in-built impls or any existing user impls.

[mjsax]

Java 11 / 2.12: kafka.server.DescribeLogDirsRequestTest.testDescribeLogDirsRequest
Java 11 / 2.13: kafka.api.ConsumerBounceTest.testRollingBrokerRestartsWithSmallerMaxGroupSizeConfigDisruptsBigGroup
Java 8: kafka.api.SaslPlainSslEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl

Retest this please.

[big-andy-coates]

retest this please

[big-andy-coates]

JDK 11 / 2.13 - test log unavailable

retest this please

[guozhangwang]

LGTM!.

[mjsax]

Java 11 / 2.12:

kafka.api.SaslGssapiSslEndToEndAuthorizationTest.testNoDescribeProduceOrConsumeWithoutTopicDescribeAcl
org.apache.kafka.streams.state.internals.RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest.shouldForwardAllDbOptionsCalls

Java 11 / 2.13: org.apache.kafka.connect.integration.RebalanceSourceConnectorsIntegrationTest.testReconfigConnector

Java 8 passed.