From 681c6fc1fb37c9cfec00df09fdfe5bf1b11f3b4b Mon Sep 17 00:00:00 2001
From: Guozhang Wang <wangguoz@gmail.com>
Date: Wed, 31 May 2017 19:30:43 -0700
Subject: [PATCH] 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
---
 .../src/main/java/org/apache/kafka/streams/KafkaStreams.java   | 2 +-
 .../src/main/java/org/apache/kafka/streams/StreamsMetrics.java | 2 +-
 .../apache/kafka/streams/errors/BrokerNotFoundException.java   | 2 --
 .../kafka/streams/errors/InvalidStateStoreException.java       | 2 --
 .../java/org/apache/kafka/streams/errors/LockException.java    | 2 --
 .../apache/kafka/streams/errors/ProcessorStateException.java   | 2 --
 .../java/org/apache/kafka/streams/errors/StreamsException.java | 2 --
 .../org/apache/kafka/streams/errors/TaskIdFormatException.java | 2 --
 .../apache/kafka/streams/errors/TopologyBuilderException.java  | 2 --
 .../main/java/org/apache/kafka/streams/kstream/Aggregator.java | 2 --
 .../java/org/apache/kafka/streams/kstream/ForeachAction.java   | 2 --
 .../java/org/apache/kafka/streams/kstream/GlobalKTable.java    | 2 +-
 .../java/org/apache/kafka/streams/kstream/Initializer.java     | 2 --
 .../java/org/apache/kafka/streams/kstream/JoinWindows.java     | 2 --
 .../java/org/apache/kafka/streams/kstream/KGroupedStream.java  | 2 +-
 .../java/org/apache/kafka/streams/kstream/KGroupedTable.java   | 2 +-
 .../main/java/org/apache/kafka/streams/kstream/KStream.java    | 2 +-
 .../java/org/apache/kafka/streams/kstream/KStreamBuilder.java  | 2 ++
 .../src/main/java/org/apache/kafka/streams/kstream/KTable.java | 2 +-
 .../java/org/apache/kafka/streams/kstream/KeyValueMapper.java  | 2 --
 .../src/main/java/org/apache/kafka/streams/kstream/Merger.java | 2 --
 .../main/java/org/apache/kafka/streams/kstream/Predicate.java  | 2 --
 .../main/java/org/apache/kafka/streams/kstream/Reducer.java    | 3 +--
 .../java/org/apache/kafka/streams/kstream/SessionWindows.java  | 2 --
 .../java/org/apache/kafka/streams/kstream/TimeWindows.java     | 2 --
 .../java/org/apache/kafka/streams/kstream/Transformer.java     | 2 --
 .../org/apache/kafka/streams/kstream/TransformerSupplier.java  | 2 --
 .../org/apache/kafka/streams/kstream/UnlimitedWindows.java     | 2 --
 .../java/org/apache/kafka/streams/kstream/ValueJoiner.java     | 2 --
 .../java/org/apache/kafka/streams/kstream/ValueMapper.java     | 2 --
 .../org/apache/kafka/streams/kstream/ValueTransformer.java     | 2 --
 .../apache/kafka/streams/kstream/ValueTransformerSupplier.java | 2 --
 .../src/main/java/org/apache/kafka/streams/kstream/Window.java | 2 --
 .../main/java/org/apache/kafka/streams/kstream/Windowed.java   | 2 --
 .../main/java/org/apache/kafka/streams/kstream/Windows.java    | 2 --
 .../streams/processor/ExtractRecordMetadataTimestamp.java      | 2 ++
 .../apache/kafka/streams/processor/FailOnInvalidTimestamp.java | 2 ++
 .../kafka/streams/processor/LogAndSkipOnInvalidTimestamp.java  | 2 ++
 .../java/org/apache/kafka/streams/processor/Processor.java     | 3 +++
 .../org/apache/kafka/streams/processor/ProcessorContext.java   | 2 +-
 .../apache/kafka/streams/processor/StateRestoreCallback.java   | 3 +++
 .../org/apache/kafka/streams/processor/TimestampExtractor.java | 2 ++
 .../org/apache/kafka/streams/processor/TopologyBuilder.java    | 2 ++
 .../streams/processor/UsePreviousTimeOnInvalidTimestamp.java   | 2 ++
 .../kafka/streams/processor/WallclockTimestampExtractor.java   | 2 ++
 .../java/org/apache/kafka/streams/state/KeyValueStore.java     | 2 --
 .../org/apache/kafka/streams/state/QueryableStoreType.java     | 2 --
 .../org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java  | 2 --
 .../org/apache/kafka/streams/state/ReadOnlySessionStore.java   | 2 --
 .../org/apache/kafka/streams/state/ReadOnlyWindowStore.java    | 2 --
 .../src/main/java/org/apache/kafka/streams/state/Stores.java   | 2 ++
 .../java/org/apache/kafka/streams/state/StreamsMetadata.java   | 2 ++
 .../main/java/org/apache/kafka/streams/state/WindowStore.java  | 2 --
 53 files changed, 35 insertions(+), 74 deletions(-)

diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
index 6da22edab16..c16f379b64b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
+++ b/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);
diff --git a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java b/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
index 773d0232e5b..d4f685b5deb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/StreamsMetrics.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/BrokerNotFoundException.java b/streams/src/main/java/org/apache/kafka/streams/errors/BrokerNotFoundException.java
index 157fbd92415..24c5fcb480e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/BrokerNotFoundException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java b/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
index a6ee1f1fb95..6f770736c62 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/InvalidStateStoreException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java b/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
index d5cf858e482..ddaa69fcb8d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/LockException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java b/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
index 1b659bf7e0b..8f1f6ac1169 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/ProcessorStateException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java b/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
index dc6b1e3640a..41c3c31a8d9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/StreamsException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java b/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
index c177705a536..63493436816 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TaskIdFormatException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java b/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
index 387ab12808a..b9c0c3a8bcb 100644
--- a/streams/src/main/java/org/apache/kafka/streams/errors/TopologyBuilderException.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java
index 61218490f93..4eec4f59197 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Aggregator.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
index 09212a9a37f..3e48d79e957 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ForeachAction.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java
index d4be41509f9..7266d7d568f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/GlobalKTable.java
+++ b/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> {
 }
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java
index 423791de864..d41c638663b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Initializer.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
index 2f50dc8e4fd..9d697382f1b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/JoinWindows.java
+++ b/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. */
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java
index 2cdf047865f..e02231afee9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedStream.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java
index 2079860ca4f..5d019c4a648 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KGroupedTable.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
index bb37af81d1d..c6da8e59b09 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStream.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
index 59b8c6fb3c0..b941f780bbc 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KStreamBuilder.java
+++ b/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);
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
index e6219c2833b..7f8ab6aafb4 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KTable.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java
index 98ada013556..e707fbb3b26 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/KeyValueMapper.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Merger.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Merger.java
index a2859458c40..6e6b01a21b6 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Merger.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java
index ac8cd1b6ba0..4c7ca0576f9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Predicate.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java
index f29b598febb..e7cc2343048 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Reducer.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
index 20334458f99..6e06461f2c1 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/SessionWindows.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
index 7a2d00e67a4..38362ada546 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TimeWindows.java
+++ b/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. */
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
index f0a322e3631..7265a11a5a7 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Transformer.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
index ba8aba9b226..b680347434a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/TransformerSupplier.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
index cf0366d0186..e116a8bbe4c 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/UnlimitedWindows.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java
index 7392cb0e4ad..1ed6003c5f9 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueJoiner.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java
index 0c1a0c96016..51396b5ff4f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueMapper.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
index 6c6ec50496b..0936e7a0c76 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformer.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java b/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
index 2adc060910b..78234aec80f 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/ValueTransformerSupplier.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
index c342112f1b0..08540a1aadf 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Window.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
index 72347973286..576706e6ff8 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Windowed.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java b/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
index d44db534635..09fdfce948d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java
+++ b/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;
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ExtractRecordMetadataTimestamp.java b/streams/src/main/java/org/apache/kafka/streams/processor/ExtractRecordMetadataTimestamp.java
index 59575f5161c..79c8dd34e54 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/ExtractRecordMetadataTimestamp.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/FailOnInvalidTimestamp.java b/streams/src/main/java/org/apache/kafka/streams/processor/FailOnInvalidTimestamp.java
index 9b90c0872aa..e8fc78c7b5a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/FailOnInvalidTimestamp.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/LogAndSkipOnInvalidTimestamp.java b/streams/src/main/java/org/apache/kafka/streams/processor/LogAndSkipOnInvalidTimestamp.java
index 228f2ee8101..0561e61011d 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/LogAndSkipOnInvalidTimestamp.java
+++ b/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);
 
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/Processor.java b/streams/src/main/java/org/apache/kafka/streams/processor/Processor.java
index b3f40d10b28..2aaf45e4615 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/Processor.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java b/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
index ea50fdc2961..559e9f76f9b 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/ProcessorContext.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreCallback.java b/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreCallback.java
index 2e896c8a015..91732c3dfab 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/StateRestoreCallback.java
+++ b/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);
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java b/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
index f5e0e1d3181..bdd44037bdd 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TimestampExtractor.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
index 08839cd0cde..f9a0397127e 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/TopologyBuilder.java
+++ b/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);
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/UsePreviousTimeOnInvalidTimestamp.java b/streams/src/main/java/org/apache/kafka/streams/processor/UsePreviousTimeOnInvalidTimestamp.java
index bc96110ee91..dd952ccf120 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/UsePreviousTimeOnInvalidTimestamp.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/WallclockTimestampExtractor.java b/streams/src/main/java/org/apache/kafka/streams/processor/WallclockTimestampExtractor.java
index 555ed105b63..ad3b3bc75de 100644
--- a/streams/src/main/java/org/apache/kafka/streams/processor/WallclockTimestampExtractor.java
+++ b/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 {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
index a6addba9c3a..bd188350f83 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/KeyValueStore.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java b/streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java
index 6383a9c0fbc..11b849b6b01 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/QueryableStoreType.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java b/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java
index c638085da83..76bb47bc40a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyKeyValueStore.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java b/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java
index 7079769c151..094e3fcc794 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlySessionStore.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java
index 51864e51d8e..b128c586221 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/ReadOnlyWindowStore.java
+++ b/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> {
 
     /**
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java b/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
index bb5abf8e80e..86ee1d25b94 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/Stores.java
+++ b/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);
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java b/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java
index 198b1b881d1..025f36c1f7a 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/StreamsMetadata.java
+++ b/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
diff --git a/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java b/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
index c9d5d7166cc..fc8ca6fae77 100644
--- a/streams/src/main/java/org/apache/kafka/streams/state/WindowStore.java
+++ b/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> {
 
     /**