MINOR: Add Functional Interface annotation to interfaces used by Lambdas (#19234)

Adds the FunctionalInterface annotation to relevant Kafka Streams
classes. While this is not strictly required for Java, it's still best
practice and also useful for better integration with other JVM
languages, for example Clojure, to allow using these interfaces as
lambdas.

Reviewers: Matthias J. Sax <matthias@confluent.io>

clients/src/main/java/org/apache/kafka/clients/consumer/OffsetCommitCallback.java

@@ -26,6 +26,7 @@ import java.util.Map;
  * A callback interface that the user can implement to trigger custom actions when a commit request completes. The callback
  * may be executed in any thread calling {@link Consumer#poll(java.time.Duration) poll()}.
  */
+@FunctionalInterface
 public interface OffsetCommitCallback {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java

@@ -38,6 +38,7 @@ package org.apache.kafka.streams.kstream;
  * @see SessionWindowedKStream#aggregate(Initializer, Aggregator, Merger, Materialized)
  * @see Reducer
  */
+@FunctionalInterface
 public interface Aggregator<K, V, VAgg> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java

@@ -30,6 +30,7 @@ package org.apache.kafka.streams.kstream;
  *
  * @see KStream#foreach(ForeachAction)
  */
+@FunctionalInterface
 public interface ForeachAction<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java

@@ -30,6 +30,7 @@ package org.apache.kafka.streams.kstream;
  * @see SessionWindowedKStream#aggregate(Initializer, Aggregator, Merger)
  * @see SessionWindowedKStream#aggregate(Initializer, Aggregator, Merger, Materialized)
  */
+@FunctionalInterface
 public interface Initializer<VAgg> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java

@@ -44,6 +44,7 @@ import org.apache.kafka.streams.KeyValue;
  * @see KTable#groupBy(KeyValueMapper, Grouped)
  * @see KTable#toStream(KeyValueMapper)
  */
+@FunctionalInterface
 public interface KeyValueMapper<K, V, VR> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/Merger.java

@@ -23,6 +23,7 @@ package org.apache.kafka.streams.kstream;
  * @param <K>   key type
  * @param <V>   aggregate value type
  */
+@FunctionalInterface
 public interface Merger<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java

@@ -31,6 +31,7 @@ import org.apache.kafka.streams.KeyValue;
  * @see KTable#filter(Predicate)
  * @see KTable#filterNot(Predicate)
  */
+@FunctionalInterface
 public interface Predicate<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java

@@ -36,6 +36,7 @@ import org.apache.kafka.streams.KeyValue;
  * @see SessionWindowedKStream#reduce(Reducer, Materialized)
  * @see Aggregator
  */
+@FunctionalInterface
 public interface Reducer<V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java

@@ -38,6 +38,7 @@ import java.util.function.Supplier;
  * @deprecated Since 4.0. Use {@link org.apache.kafka.streams.processor.api.ProcessorSupplier api.ProcessorSupplier} instead.
  */
 @Deprecated
+@FunctionalInterface
 public interface TransformerSupplier<K, V, R> extends ConnectedStoreProvider, Supplier<Transformer<K, V, R>> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java

@@ -40,6 +40,7 @@ package org.apache.kafka.streams.kstream;
  * @see KTable#leftJoin(KTable, ValueJoiner)
  * @see KTable#outerJoin(KTable, ValueJoiner)
  */
+@FunctionalInterface
 public interface ValueJoiner<V1, V2, VR> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoinerWithKey.java

@@ -44,6 +44,7 @@ package org.apache.kafka.streams.kstream;
  * @see KStream#leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey)
  * @see KStream#leftJoin(GlobalKTable, KeyValueMapper, ValueJoinerWithKey, Named)
  */
+@FunctionalInterface
 public interface ValueJoinerWithKey<K1, V1, V2, VR> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java

@@ -37,6 +37,7 @@ import org.apache.kafka.streams.processor.api.FixedKeyProcessor;
  * @see KTable#mapValues(ValueMapper)
  * @see KTable#mapValues(ValueMapperWithKey)
  */
+@FunctionalInterface
 public interface ValueMapper<V, VR> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapperWithKey.java

@@ -38,6 +38,7 @@ package org.apache.kafka.streams.kstream;
  * @see KTable#mapValues(ValueMapperWithKey)
  */
 
+@FunctionalInterface
 public interface ValueMapperWithKey<K, V, VR> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java

@@ -36,6 +36,7 @@ import org.apache.kafka.streams.processor.api.FixedKeyProcessorSupplier;
  * @deprecated Since 4.0. Use {@link FixedKeyProcessorSupplier} instead.
  */
 @Deprecated
+@FunctionalInterface
 public interface ValueTransformerSupplier<V, VR> extends ConnectedStoreProvider {
 
     /**

streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerWithKeySupplier.java

@@ -35,6 +35,7 @@ import java.util.function.Supplier;
  * @see Transformer
  * @see TransformerSupplier
  */
+@FunctionalInterface
 public interface ValueTransformerWithKeySupplier<K, V, VR> extends ConnectedStoreProvider, Supplier<ValueTransformerWithKey<K, V, VR>> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/BatchingStateRestoreCallback.java

@@ -28,6 +28,7 @@ import java.util.Collection;
  * It is expected that implementations of this class will not call the {@link StateRestoreCallback#restore(byte[],
  * byte[])} method.
  */
+@FunctionalInterface
 public interface BatchingStateRestoreCallback extends StateRestoreCallback {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/Cancellable.java

@@ -23,6 +23,7 @@ import java.time.Duration;
  *
  * @see Punctuator
  */
+@FunctionalInterface
 public interface Cancellable {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/Punctuator.java

@@ -26,6 +26,7 @@ import java.time.Duration;
  *
  * @see Cancellable
  */
+@FunctionalInterface
 public interface Punctuator {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreCallback.java

@@ -20,6 +20,7 @@ package org.apache.kafka.streams.processor;
  * Restoration logic for log-backed state stores upon restart,
  * it takes one record at a time from the logs to apply to the restoring state.
  */
+@FunctionalInterface
 public interface StateRestoreCallback {
 
     void restore(byte[] key, byte[] value);

streams/src/main/java/org/apache/kafka/streams/processor/StreamPartitioner.java

@@ -50,6 +50,7 @@ import java.util.Set;
  *      org.apache.kafka.common.serialization.Serializer, StreamPartitioner, String...)
  * @see Topology#addSink(String, String, StreamPartitioner, String...)
  */
+@FunctionalInterface
 public interface StreamPartitioner<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java

@@ -23,6 +23,7 @@ import org.apache.kafka.streams.kstream.KTable;
  * An interface that allows the Kafka Streams framework to extract a timestamp from an instance of {@link ConsumerRecord}.
  * The extracted timestamp is defined as milliseconds.
  */
+@FunctionalInterface
 public interface TimestampExtractor {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/TopicNameExtractor.java

@@ -19,6 +19,7 @@ package org.apache.kafka.streams.processor;
 /**
  * An interface that allows to dynamically determine the name of the Kafka topic to send at the sink node of the topology.
  */
+@FunctionalInterface
 public interface TopicNameExtractor<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/api/FixedKeyProcessor.java

@@ -29,6 +29,7 @@ import java.time.Duration;
  * @param <VIn> the type of input values
  * @param <VOut> the type of output values
  */
+@FunctionalInterface
 public interface FixedKeyProcessor<KIn, VIn, VOut> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/api/Processor.java

@@ -30,6 +30,7 @@ import java.time.Duration;
  * @param <KOut> the type of output keys
  * @param <VOut> the type of output values
  */
+@FunctionalInterface
 public interface Processor<KIn, VIn, KOut, VOut> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/processor/api/WrappedFixedKeyProcessorSupplier.java

@@ -24,6 +24,7 @@ package org.apache.kafka.streams.processor.api;
  * To convert a {@link FixedKeyProcessorSupplier} instance into a {@link WrappedFixedKeyProcessorSupplier},
  * use the {@link ProcessorWrapper#asWrappedFixedKey(FixedKeyProcessorSupplier)} method
  */
+@FunctionalInterface
 public interface WrappedFixedKeyProcessorSupplier<KIn, VIn, VOut> extends FixedKeyProcessorSupplier<KIn, VIn, VOut> {
 
 }

streams/src/main/java/org/apache/kafka/streams/processor/api/WrappedProcessorSupplier.java

@@ -24,6 +24,7 @@ package org.apache.kafka.streams.processor.api;
  * To convert a {@link ProcessorSupplier} instance into a {@link WrappedProcessorSupplier},
  * use the {@link ProcessorWrapper#asWrapped(ProcessorSupplier)} method
  */
+@FunctionalInterface
 public interface WrappedProcessorSupplier<KIn, VIn, KOut, VOut> extends ProcessorSupplier<KIn, VIn, KOut, VOut> {
 
 }