KAFKA-5350: Modify unstable annotations in Streams API

Author: Guozhang Wang <wangguoz@gmail.com>

Reviewers: Matthias J. Sax <matthias@confluent.io>, Jason Gustafson <jason@confluent.io>

Closes #3172 from guozhangwang/K5350-compatibility-annotations

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

@@ -115,7 +115,7 @@ import static org.apache.kafka.streams.StreamsConfig.PROCESSING_GUARANTEE_CONFIG
  * @see KStreamBuilder
  * @see TopologyBuilder
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public class KafkaStreams {
 
     private static final Logger log = LoggerFactory.getLogger(KafkaStreams.class);

streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java

@@ -26,7 +26,7 @@ import java.util.Map;
 /**
  * The Kafka Streams metrics interface for adding metric sensors and collecting metric values.
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface StreamsMetrics {
 
     /**

streams/src/main/java/org/apache/kafka/streams/errors/BrokerNotFoundException.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates that none of the specified {@link org.apache.kafka.streams.StreamsConfig#BOOTSTRAP_SERVERS_CONFIG brokers}
@@ -24,7 +23,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  *
  * @see org.apache.kafka.streams.StreamsConfig
  */
-@InterfaceStability.Unstable
 public class BrokerNotFoundException extends StreamsException {
 
     private final static long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates that there was a problem when trying to access a
@@ -27,7 +26,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * being (re)-initialized. Once the rebalance has completed the stores will be available again. Hence, it is valid
  * to backoff and retry when handling this exception.
  */
-@InterfaceStability.Unstable
 public class InvalidStateStoreException extends StreamsException {
 
     private final static long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/LockException.java

@@ -16,14 +16,12 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates that the state store directory lock could not be acquired because another thread holds the lock.
  *
  * @see org.apache.kafka.streams.processor.StateStore
  */
-@InterfaceStability.Unstable
 public class LockException extends StreamsException {
 
     private final static long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java

@@ -16,14 +16,12 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates a processor state operation (e.g. put, get) has failed.
  *
  * @see org.apache.kafka.streams.processor.StateStore
  */
-@InterfaceStability.Unstable
 public class ProcessorStateException extends StreamsException {
 
     private final static long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java

@@ -17,12 +17,10 @@
 package org.apache.kafka.streams.errors;
 
 import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * {@link StreamsException} is the top-level exception type generated by Kafka Streams.
  */
-@InterfaceStability.Unstable
 public class StreamsException extends KafkaException {
 
     private final static long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates a run time error incurred while trying parse the {@link org.apache.kafka.streams.processor.TaskId task id}
@@ -24,7 +23,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  *
  * @see org.apache.kafka.streams.processor.internals.StreamTask
  */
-@InterfaceStability.Unstable
 public class TaskIdFormatException extends StreamsException {
 
     private static final long serialVersionUID = 1L;

streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java

@@ -16,13 +16,11 @@
  */
 package org.apache.kafka.streams.errors;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Indicates a pre-run time error incurred while parsing the {@link org.apache.kafka.streams.processor.TopologyBuilder
  * builder} to construct the {@link org.apache.kafka.streams.processor.internals.ProcessorTopology processor topology}.
  */
-@InterfaceStability.Unstable
 public class TopologyBuilderException extends StreamsException {
 
     private static final long serialVersionUID = 1L;

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The {@code Aggregator} interface for aggregating values of the given key.
@@ -38,7 +37,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see Reducer
  */
-@InterfaceStability.Unstable
 public interface Aggregator<K, V, VA> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The {@code ForeachAction} interface for performing an action on a {@link org.apache.kafka.streams.KeyValue key-value
@@ -31,7 +30,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KStream#foreach(ForeachAction)
  * @see KTable#foreach(ForeachAction)
  */
-@InterfaceStability.Unstable
 public interface ForeachAction<K, V> {
 
     /**

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

@@ -64,6 +64,6 @@ import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  * @see KStream#join(GlobalKTable, KeyValueMapper, ValueJoiner)
  * @see KStream#leftJoin(GlobalKTable, KeyValueMapper, ValueJoiner)
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface GlobalKTable<K, V> {
 }

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The {@code Initializer} interface for creating an initial value in aggregations.
@@ -31,7 +30,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, String)
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, org.apache.kafka.streams.processor.StateStoreSupplier)
  */
-@InterfaceStability.Unstable
 public interface Initializer<VA> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 import java.util.Map;
@@ -63,7 +62,6 @@ import java.util.Map;
  * @see KStream#outerJoin(KStream, ValueJoiner, JoinWindows)
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public final class JoinWindows extends Windows<Window> {
 
     /** Maximum time difference for tuples that are before the join tuple. */

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

@@ -42,7 +42,7 @@ import org.apache.kafka.streams.state.WindowStore;
  * @param <V> Type of values
  * @see KStream
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface KGroupedStream<K, V> {
 
     /**

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

@@ -38,7 +38,7 @@ import org.apache.kafka.streams.state.QueryableStoreType;
  * @param <V> Type of values
  * @see KTable
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface KGroupedTable<K, V> {
 
     /**

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

@@ -54,7 +54,7 @@ import org.apache.kafka.streams.processor.TopologyBuilder;
  * @see KStreamBuilder#stream(String...)
  */
 @SuppressWarnings("unused")
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface KStream<K, V> {
 
     /**

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.kstream;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.streams.KafkaStreams;
@@ -47,6 +48,7 @@ import java.util.regex.Pattern;
  * @see KTable
  * @see GlobalKTable
  */
+@InterfaceStability.Evolving
 public class KStreamBuilder extends TopologyBuilder {
 
     private final AtomicInteger index = new AtomicInteger(0);

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

@@ -64,7 +64,7 @@ import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
  * @see GlobalKTable
  * @see KStreamBuilder#table(String, String)
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface KTable<K, V> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 
 /**
@@ -45,7 +44,6 @@ import org.apache.kafka.streams.KeyValue;
  * @see KTable#groupBy(KeyValueMapper, org.apache.kafka.common.serialization.Serde, org.apache.kafka.common.serialization.Serde)
  * @see KTable#toStream(KeyValueMapper)
  */
-@InterfaceStability.Unstable
 public interface KeyValueMapper<K, V, VR> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The interface for merging aggregate values for {@link SessionWindows} with the given key.
@@ -24,7 +23,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @param <K>   key type
  * @param <V>   aggregate value type
  */
-@InterfaceStability.Unstable
 public interface Merger<K, V> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 
 /**
@@ -32,7 +31,6 @@ import org.apache.kafka.streams.KeyValue;
  * @see KTable#filter(Predicate)
  * @see KTable#filterNot(Predicate)
  */
-@InterfaceStability.Unstable
 public interface Predicate<K, V> {
 
     /**

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

@@ -16,8 +16,8 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
+
 /**
  * The {@code Reducer} interface for combining two values of the same type into a new value.
  * In contrast to {@link Aggregator} the result type must be the same as the input type.
@@ -36,7 +36,6 @@ import org.apache.kafka.streams.KeyValue;
  * @see KGroupedStream#reduce(Reducer, SessionWindows, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see Aggregator
  */
-@InterfaceStability.Unstable
 public interface Reducer<V> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 /**
@@ -67,7 +66,6 @@ import org.apache.kafka.streams.processor.TimestampExtractor;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public final class SessionWindows {
 
     private final long gapMs;

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.kstream.internals.TimeWindow;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
@@ -54,7 +53,6 @@ import java.util.Map;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Windows, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public final class TimeWindows extends Windows<TimeWindow> {
 
     /** The size of the windows in milliseconds. */

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.processor.ProcessorContext;
 import org.apache.kafka.streams.processor.StateStore;
@@ -45,7 +44,6 @@ import org.apache.kafka.streams.processor.TimestampExtractor;
  * @see KStream#map(KeyValueMapper)
  * @see KStream#flatMap(KeyValueMapper)
  */
-@InterfaceStability.Unstable
 public interface Transformer<K, V, R> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * A {@code TransformerSupplier} interface which can create one or more {@link Transformer} instances.
@@ -31,7 +30,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see ValueTransformerSupplier
  * @see KStream#transformValues(ValueTransformerSupplier, String...)
  */
-@InterfaceStability.Unstable
 public interface TransformerSupplier<K, V, R> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.kstream.internals.UnlimitedWindow;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
@@ -43,7 +42,6 @@ import java.util.Map;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Windows, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public final class UnlimitedWindows extends Windows<UnlimitedWindow> {
 
     private static final long DEFAULT_START_TIMESTAMP_MS = 0L;

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The {@code ValueJoiner} interface for joining two values into a new value of arbitrary type.
@@ -41,7 +40,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KTable#leftJoin(KTable, ValueJoiner)
  * @see KTable#outerJoin(KTable, ValueJoiner)
  */
-@InterfaceStability.Unstable
 public interface ValueJoiner<V1, V2, VR> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The {@code ValueMapper} interface for mapping a value to a new value of arbitrary type.
@@ -34,7 +33,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KStream#flatMapValues(ValueMapper)
  * @see KTable#mapValues(ValueMapper)
  */
-@InterfaceStability.Unstable
 public interface ValueMapper<V, VR> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.errors.StreamsException;
 import org.apache.kafka.streams.processor.ProcessorContext;
@@ -42,7 +41,6 @@ import org.apache.kafka.streams.processor.TimestampExtractor;
  * @see KStream#transformValues(ValueTransformerSupplier, String...)
  * @see Transformer
  */
-@InterfaceStability.Unstable
 public interface ValueTransformer<V, VR> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * A {@code ValueTransformerSupplier} interface which can create one or more {@link ValueTransformer} instances.
@@ -29,7 +28,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see TransformerSupplier
  * @see KStream#transform(TransformerSupplier, String...)
  */
-@InterfaceStability.Unstable
 public interface ValueTransformerSupplier<V, VR> {
 
     /**

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 /**
@@ -33,7 +32,6 @@ import org.apache.kafka.streams.processor.TimestampExtractor;
  * @see org.apache.kafka.streams.kstream.internals.UnlimitedWindow
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public abstract class Window {
 
     protected final long startMs;

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * The result key type of a windowed stream aggregation.
@@ -40,7 +39,6 @@ import org.apache.kafka.common.annotation.InterfaceStability;
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, org.apache.kafka.streams.processor.StateStoreSupplier)
  * @see KGroupedStream#aggregate(Initializer, Aggregator, Merger, SessionWindows, org.apache.kafka.common.serialization.Serde, String)
  */
-@InterfaceStability.Unstable
 public class Windowed<K> {
 
     private final K key;

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

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.kstream;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.TimestampExtractor;
 
 import java.util.Map;
@@ -35,7 +34,6 @@ import java.util.Map;
  * @see SessionWindows
  * @see TimestampExtractor
  */
-@InterfaceStability.Unstable
 public abstract class Windows<W extends Window> {
 
     private static final int DEFAULT_NUM_SEGMENTS = 3;

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Retrieves embedded metadata timestamps from Kafka messages.
@@ -42,6 +43,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
  * @see UsePreviousTimeOnInvalidTimestamp
  * @see WallclockTimestampExtractor
  */
+@InterfaceStability.Evolving
 abstract class ExtractRecordMetadataTimestamp implements TimestampExtractor {
 
     /**

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.errors.StreamsException;
 
 /**
@@ -42,6 +43,7 @@ import org.apache.kafka.streams.errors.StreamsException;
  * @see UsePreviousTimeOnInvalidTimestamp
  * @see WallclockTimestampExtractor
  */
+@InterfaceStability.Evolving
 public class FailOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
 
     /**

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,6 +46,7 @@ import org.slf4j.LoggerFactory;
  * @see UsePreviousTimeOnInvalidTimestamp
  * @see WallclockTimestampExtractor
  */
+@InterfaceStability.Evolving
 public class LogAndSkipOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
     private static final Logger log = LoggerFactory.getLogger(LogAndSkipOnInvalidTimestamp.class);
 

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

@@ -16,12 +16,15 @@
  */
 package org.apache.kafka.streams.processor;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
  * A processor of key-value pair records.
  *
  * @param <K> the type of keys
  * @param <V> the type of values
  */
+@InterfaceStability.Evolving
 public interface Processor<K, V> {
 
     /**

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

@@ -26,7 +26,7 @@ import java.util.Map;
 /**
  * Processor context interface.
  */
-@InterfaceStability.Unstable
+@InterfaceStability.Evolving
 public interface ProcessorContext {
 
     /**

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

@@ -16,10 +16,13 @@
  */
 package org.apache.kafka.streams.processor;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
+
 /**
  * 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.
  */
+@InterfaceStability.Evolving
 public interface StateRestoreCallback {
 
     void restore(byte[] key, byte[] value);

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

@@ -17,12 +17,14 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 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.
  */
+@InterfaceStability.Evolving
 public interface TimestampExtractor {
 
     /**

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

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.processor;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.Serializer;
 import org.apache.kafka.common.utils.Utils;
@@ -60,6 +61,7 @@ import java.util.regex.Pattern;
  * to construct an acyclic graph of these nodes, and the builder is then passed into a new {@link org.apache.kafka.streams.KafkaStreams}
  * instance that will then {@link org.apache.kafka.streams.KafkaStreams#start() begin consuming, processing, and producing records}.
  */
+@InterfaceStability.Evolving
 public class TopologyBuilder {
 
     private static final Logger log = LoggerFactory.getLogger(TopologyBuilder.class);

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.errors.StreamsException;
 
 /**
@@ -42,6 +43,7 @@ import org.apache.kafka.streams.errors.StreamsException;
  * @see LogAndSkipOnInvalidTimestamp
  * @see WallclockTimestampExtractor
  */
+@InterfaceStability.Evolving
 public class UsePreviousTimeOnInvalidTimestamp extends ExtractRecordMetadataTimestamp {
 
     /**

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

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.processor;
 
 import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.annotation.InterfaceStability;
 
 /**
  * Retrieves current wall clock timestamps as {@link System#currentTimeMillis()}.
@@ -30,6 +31,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord;
  * @see LogAndSkipOnInvalidTimestamp
  * @see UsePreviousTimeOnInvalidTimestamp
  */
+@InterfaceStability.Evolving
 public class WallclockTimestampExtractor implements TimestampExtractor {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.processor.StateStore;
 
@@ -28,7 +27,6 @@ import java.util.List;
  * @param <K> The key type
  * @param <V> The value type
  */
-@InterfaceStability.Unstable
 public interface KeyValueStore<K, V> extends StateStore, ReadOnlyKeyValueStore<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.StateStore;
 import org.apache.kafka.streams.state.internals.StateStoreProvider;
 
@@ -28,7 +27,6 @@ import org.apache.kafka.streams.state.internals.StateStoreProvider;
  *
  * @param <T>   The store type
  */
-@InterfaceStability.Unstable
 public interface QueryableStoreType<T> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.errors.InvalidStateStoreException;
 
 /**
@@ -27,7 +26,6 @@ import org.apache.kafka.streams.errors.InvalidStateStoreException;
  * @param <K> the key type
  * @param <V> the value type
  */
-@InterfaceStability.Unstable
 public interface ReadOnlyKeyValueStore<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java

@@ -17,7 +17,6 @@
 package org.apache.kafka.streams.state;
 
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.kstream.Windowed;
 
 /**
@@ -28,7 +27,6 @@ import org.apache.kafka.streams.kstream.Windowed;
  * @param <K> the key type
  * @param <AGG> the aggregated value type
  */
-@InterfaceStability.Unstable
 public interface ReadOnlySessionStore<K, AGG> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.errors.InvalidStateStoreException;
 import org.apache.kafka.streams.kstream.Windowed;
 
@@ -27,7 +26,6 @@ import org.apache.kafka.streams.kstream.Windowed;
  * @param <K> Type of keys
  * @param <V> Type of values
  */
-@InterfaceStability.Unstable
 public interface ReadOnlyWindowStore<K, V> {
 
     /**

streams/src/main/java/org/apache/kafka/streams/state/Stores.java

@@ -16,6 +16,7 @@
  */
 package org.apache.kafka.streams.state;
 
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.processor.StateStoreSupplier;
@@ -34,6 +35,7 @@ import java.util.Map;
 /**
  * Factory for creating state stores in Kafka Streams.
  */
+@InterfaceStability.Evolving
 public class Stores {
 
     private static final Logger log = LoggerFactory.getLogger(Stores.class);

streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.streams.state;
 
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.KafkaStreams;
 
 import java.util.Collections;
@@ -29,6 +30,7 @@ import java.util.Set;
  * the instance and the Set of {@link TopicPartition}s available on the instance.
  * NOTE: This is a point in time view. It may change when rebalances happen.
  */
+@InterfaceStability.Evolving
 public class StreamsMetadata {
     /**
      * Sentinel to indicate that the StreamsMetadata is currently unavailable. This can occur during rebalance

streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java

@@ -16,7 +16,6 @@
  */
 package org.apache.kafka.streams.state;
 
-import org.apache.kafka.common.annotation.InterfaceStability;
 import org.apache.kafka.streams.processor.StateStore;
 
 /**
@@ -25,7 +24,6 @@ import org.apache.kafka.streams.processor.StateStore;
  * @param <K> Type of keys
  * @param <V> Type of values
  */
-@InterfaceStability.Unstable
 public interface WindowStore<K, V> extends StateStore, ReadOnlyWindowStore<K, V> {
 
     /**