Status

Current state: Under Discussion

Discussion thread: TDB

JIRA: N/A

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

Note this KIP is the original work of Ray McDermott -  Due to lack of self-service signup for wiki ID and editing permissions, we're adding Ray's text to this KIP.

Motivation

Jackdaw is a Clojure library over the Kafka Streams DSL. In its current state, Jackdaw uses wrappers around the DSL interfaces to enable function passing in Clojure as one would enjoy if using Java.

Such wrappers are verbose and add to complexity to the library. They would not be required if Clojure has first class interop for SAM interfaces.

The new Clojure version 1.12 has many updates to Java interop including SAM interfaces. Unfortunately, it does not quite deliver what we need with respect to thinning down Kafka Streams interop.

We were specifically hoping that passing (fn [] ...) to SAM interfaces would just work and thus we would no longer need to wrap the interface. This feature may eventually be added to the language but it currently only works for interfaces that have been explicitly annotated with @FunctionalInterface. See this JIRA ticket for details.

Java does not require the @FunctionalInterface annotation although it clarifies the intended interface purpose.

The Kafka Streams DSL does not have those annotations and this KIP proposes to add them.

Public Interfaces

No new interfaces or changes to the behavior of existing interfaces will be introduced in this KIP.

These annotations on the interfaces are new but do not affect client calls.  Details of adding the annotation are in the next section.

Proposed Changes

Add @FunctionalInterface to the Kafka Streams DSL and the Processor API (PAPI).

The interface must conform to these rules:

• have one abstract method
  • can have any number of default methods
  • can have any number of methods directly on Object

The details of each included / excluded interface is as follows:

DSL (org.apache.kafka.streams.kstream)

Interfaces in scope

We will add @FunctionalInterface to these interfaces:

• Aggregator
  • apply is the only method in the interface (SAM)
• ForeachAction
  • apply is the only method in the interface (SAM)
• GlobalKTable
  • queryableStoreName is the only method in the interface (SAM)
• Initializer
  • apply is the only method in the interface (SAM)
• KeyValueMapper
  • apply is the only method in the interface (SAM)
• Merger
  • apply is the only method in the interface (SAM)
• NamedOperation
  • withName is the only method in the interface (SAM)
• Predicate
  • test is the only method in the interface (SAM)
• Reducer
  • apply is the only method in the interface (SAM)
• TransformerSupplier
  • get is the only method in the interface (SAM)
• ValueJoiner
  • apply is the only method in the interface (SAM)
• ValueJoinerWithKey
  • apply is the only method in the interface (SAM)
• ValueMapper
  • apply is the only method in the interface (SAM)
• ValueMapperWithKey
  • apply is the only method in the interface (SAM)
• ValueTransformerSupplier
  • get is the only method in the interface (SAM)
• ValueTransformerWithKeySupplier
  • get is the only method in the interface (SAM)

Interfaces out of scope

We will not add it to these interfaces:

• BranchedKStream
  • several, non-default methods so does not qualify.
• CogroupedKStream
  • several, non-default methods so does not qualify.
• EmitStrategy
  • several, non-default methods so does not qualify.
• KGroupedStream
  • several, non-default methods so does not qualify.
• KGroupedTable
  • several, non-default methods so does not qualify.
• KStream
  • several, non-default methods so does not qualify.
• KTable
  • several, non-default methods so does not qualify.
• SessionWindowedCogroupedKStream
  • aggregate is overloaded so does not qualify.
• SessionWindowedKStream
  • several, non-default methods so does not qualify.
• Suppressed
  • several, non-default methods so does not qualify.
• TimeWindowedCogroupedKStream
  • aggregate is overloaded so does not qualify.
• TimeWindowedKStream
  • several, non-default methods so does not qualify.
• Transformer
  • several, non-default methods so does not qualify.
• ValueTransformer
  • several, non-default methods so does not qualify.
• ValueTransformerWithKey
  • several, non-default methods so does not qualify.

PAPI (org.apache.kafka.streams.processor.api)

The PAPI already has two interfaces that have @FunctionalInterface which will be unaffected:

• FixedKeyProcessorSupplier
• ProcessorSupplier.

Interfaces in scope

We will add @FunctionalInterface to these interfaces:

• FixedKeyProcessor
  • process is the only non-default method
• Processor
  • process is the only non-default method

Interfaces out of scope

We will not add it to these interfaces:

• FixedKeyProcessorContext
  • forward is overloaded so does not qualify.
• ProcessingContext
  • several, non-default methods so does not qualify.
• ProcessorContext
  • forward is overloaded so does not qualify.
• RecordMetadata
  • several, non-default methods so does not qualify.

Compatibility, Deprecation, and Migration Plan

None required

Test Plan

Running the existing tests will be sufficient.

Rejected Alternatives

Leave off the @FunctionalInterface annotation: limits the benefits to non-Java, JVM languages.