Status

Current state: Under Discussion

Discussion thread: TBD

JIRA: KAFKA-17485 - Getting issue details... STATUS

Released: TBD (target 4.0)

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

Kafka Streams supports the KafkaClientSupplier interface, that allows users to create consumer, producer, and admin client instances, which the Kafka Streams runtime will use. With KIP-1071 (and other ideas to refactor and improve the Kafka Streams runtime internally), this model does not work any longer, because we want to move the clients and Kafka Streams code bases closer to each other, allowing Kafka Streams to build on internal client APIs (for KIP-1071 in particular, we aim to build on internal KafkaConsumer APIs). Mid to long term, we aim to more and more co-develop clients and Kafka Streams, and change Kafka Streams to not use clients as a block box as we do right now, but rather build on lower level primitives, effectively transforming Kafka Streams into a "processing client" by itself. This model required that the Kafka Streams runtime can rely on client internals, which we cannot guarantee if an externally create client instance is used.

Many developers use KafkaClientSupplier interface to wrap the standard KafkaAdminClient, KafkaConsumer, and KafkaProducer. To preserve this functionality, we propose to add a new KafkaClientInteceptor interface.

Public Interfaces

We propose to deprecate KafkaClientSupplier interface, and all related public API (eg, KafkaStreams constructors and the corresponding default.client.supplier  configuration). Additionally, we propose to add KafkaClientInterceptor interface and new KafkaStreams constructors. (We could also add a config, but it's unclear if it would be needed, so for now, we omit it.)

package org.apache.kafka.streams;

@Deprecated
interface KafkaClientSupplier { ... }

// new interface
interface KafkaClientInterceptor {

    default Admin wrapAdminClient(final KafkaAdminClient adminClient) {
        return adminClient;
    }

    default Consumer<byte[], byte[]> wrapMainConsumer(final KafkaConsumer<byte[], byte[]> mainConsumer) {
        return mainConsumer;
    }

    default Consumer<byte[], byte[]> wrapRestoreConsumer(final KafkaConsumer<byte[], byte[]> restoreConsumer) {
        return restoreConsumer;
    }

    default Consumer<byte[], byte[]> wrapGlobalConsumer(final KafkaConsumer<byte[], byte[]> globalConsumer) {
        return globalConsumer;
    }

    default Producer<byte[], byte[]> wrapProducer(final KafkaProducer<byte[], byte[]> producer) {
        return producer;
    }
}

public class KafkaStreams {

    @Deprecate
    public KafkaStreams(Topology topology, Properties props, KafkaClientSupplier clientSupplier);
    @Deprecate
    public KafkaStreams(Topology topology, Properties props, KafkaClientSupplier clientSupplier, Time time);
    @Deprecate
    public KafkaStreams(Topology topology, StreamsConfig applicationConfigs, KafkaClientSupplier clientSupplier);
    // there is no 4th overload taking `Topology, StreamsConfig, KafkaClientSupplier, Time`

    // new constructors
    public KafkaStreams(Topology topology, Properties props, KafkaClientInterceptor clientInterceptor);
    public KafkaStreams(Topology topology, Properties props, KafkaClientInterceptor clientInterceptor, Time time);
    public KafkaStreams(Topology topology, StreamsConfig applicationConfigs, KafkaClientInterceptor clientInterceptor);
    // add missing 4th overload
    public KafkaStreams(Topology topology, StreamsConfig applicationConfigs, KafkaClientInterceptor clientInterceptor, Time time);
}

public class StreamsConfig {
    @Deprecated
    public static final String DEFAULT_CLIENT_SUPPLIER_CONFIG = "default.client.supplier";

    @Deprecated
    public KafkaClientSupplier getKafkaClientSupplier();
}

Proposed Changes

We propose to replace KafkaClientSupplier with KafkaClientInterceptor interface. The new interface will get a client instance that the Kafka Streams runtime creates as parameter, and can return a wrapped client. It would be invalid to create a new client instance (and it might actually not be easily possible to create one, as we don't pass in the config Map as we do for the exiting interface). The default implementation of the "wrapper methods" just return the passed in client instance without wrapping it.

With KIP-1071 enabled, it would not be allowed to use KafkaClientSupplier any longer (as long as KafkaClientSupplier is not removed). This implies that existing Kafka Streams applications which use the KafkaClientSupplier right now, would need to switch to the new KafkaClientInterceptor first, before they can enable KIP-1071. It's also not allowed to use both the old KafkaClientSupplier and new KafkaClientInterceptor at the same time; we would raise a config exception for this case.

Compatibility, Deprecation, and Migration Plan

We deprecate exiting APIs which is not a breaking change (subject to removal, most likely in 5.0 release). The new interface mimics the functionality of the deprecated one, and thus, there is not concern with removed functionality. The only theoretically possible feature gap would be, if the KafkaClientSupplier would be used to instantiate some third party implementation of the Admin, Consumer, or Producer interfaces. We are not aware of any third party Java implementations of these interfaces that would be used in combination with Kafka Streams, and don't see this as a risk. Of course, we might not be aware of all user cases, and hope that the deprecation period of the KafkaClientSupplier interface would bring any such use-case to our attention, so we can work on other solutions if necessary.

TODO (if KIP is accepted) For Kafka Streams deprecation, please add a corresponding sub-task to KAFKA-16337 - Getting issue details... STATUS for tracking, after the KIP was approved, and include a link to the created ticket in this section.

Test Plan

Standard unit testing is sufficient for this feature.

Documentation Plan

The documentation will be updates as always, including examples, and the Kafka Streams configuration page.

Rejected Alternatives

For KIP-1071 in particular, it might be possible to add new APIs to the Consumer interface directly, which are required to tap into consumer internal (or introduce some new StreamsConsumer extends Consumer interface for this purpose, to not leak Kafka Streams specific thing into the public consumer API). However, the surface area of required methods and classes to make this work for KIP-1071 is rather high, and thus it's not desired to follow this approach. Furthermore, adding new public APIs does also not align with our mid to long term goal to move off the current model to use the clients as black boxes, and to transform Kafka Streams into a "processing client". Thus, we reject this alternative as a potential step backward.