<!--
 Licensed to the Apache Software Foundation (ASF) under one or more
 contributor license agreements.  See the NOTICE file distributed with
 this work for additional information regarding copyright ownership.
 The ASF licenses this file to You under the Apache License, Version 2.0
 (the "License"); you may not use this file except in compliance with
 the License.  You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.
-->

<script><!--#include virtual="../js/templateData.js" --></script>

<script id="content-template" type="text/x-handlebars-template">
    <h1>Upgrade Guide and API Changes</h1>
    <div class="sub-nav-sticky">
        <div class="sticky-top">
            <div style="height:35px">
                <a href="/{{version}}/documentation/streams/">Introduction</a>
                <a href="/{{version}}/documentation/streams/quickstart">Run Demo App</a>
                <a href="/{{version}}/documentation/streams/tutorial">Tutorial: Write App</a>
                <a href="/{{version}}/documentation/streams/core-concepts">Concepts</a>
                <a href="/{{version}}/documentation/streams/architecture">Architecture</a>
                <a href="/{{version}}/documentation/streams/developer-guide/">Developer Guide</a>
                <a class="active-menu-item" href="/{{version}}/documentation/streams/upgrade-guide">Upgrade</a>
            </div>
        </div>
    </div>

    <p>
        Upgrading from any older version to {{fullDotVersion}} is possible: (1) if you are upgrading from 2.0.x to {{fullDotVersion}} then a single rolling bounce is needed to swap in the new jar,
        (2) if you are upgrading from older versions than 2.0.x in the online mode, you would need two rolling bounces where
        the first rolling bounce phase you need to set config <code>upgrade.from="older version"</code> (possible values are <code>"0.10.0", "0.10.1", "0.10.2", "0.11.0", "1.0", and "1.1"</code>)
        (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>):
    </p>
    <ul>
        <li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to the version from which it is being upgrade.</li>
        <li> bounce each instance of your application once </li>
        <li> prepare your newly deployed {{fullDotVersion}} application instances for a second round of rolling bounces; make sure to remove the value for config <code>upgrade.mode</code> </li>
        <li> bounce each instance of your application once more to complete the upgrade </li>
    </ul>
    <p> As an alternative, an offline upgrade is also possible. Upgrading from any versions as old as 0.10.0.x to {{fullDotVersion}} in offline mode require the following steps: </p>
    <ul>
        <li> stop all old (e.g., 0.10.0.x) application instances </li>
        <li> update your code and swap old code and jar file with new code and new jar file </li>
        <li> restart all new ({{fullDotVersion}}) application instances </li>
    </ul>

    <p>
        Note, that a brokers must be on version 0.10.1 or higher to run a Kafka Streams application version 0.10.1 or higher;
        On-disk message format must be 0.10 or higher to run a Kafka Streams application version 1.0 or higher.
        For Kafka Streams 0.10.0, broker version 0.10.0 or higher is required.
    </p>

    <p>
        Another important thing to keep in mind: in deprecated <code>KStreamBuilder</code> class, when a <code>KTable</code> is created from a source topic via <code>KStreamBuilder.table()</code>, its materialized state store
        will reuse the source topic as its changelog topic for restoring, and will disable logging to avoid appending new updates to the source topic; in the <code>StreamsBuilder</code> class introduced in 1.0, this behavior was changed
        accidentally: we still reuse the source topic as the changelog topic for restoring, but will also create a separate changelog topic to append the update records from source topic to. In the 2.0 release, we have fixed this issue and now users
        can choose whether or not to reuse the source topic based on the <code>StreamsConfig#TOPOLOGY_OPTIMIZATION</code>: if you are upgrading from the old <code>KStreamBuilder</code> class and hence you need to change your code to use
        the new <code>StreamsBuilder</code>, you should set this config value to <code>StreamsConfig#OPTIMIZE</code> to continue reusing the source topic; if you are upgrading from 1.0 or 1.1 where you are already using <code>StreamsBuilder</code> and hence have already
        created a separate changelog topic, you should set this config value to <code>StreamsConfig#NO_OPTIMIZATION</code> when upgrading to {{fullDotVersion}} in order to use that changelog topic for restoring the state store.
        More details about the new config <code>StreamsConfig#TOPOLOGY_OPTIMIZATION</code> can be found in <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-295%3A+Add+Streams+Configuration+Allowing+for+Optional+Topology+Optimization">KIP-295</a>.
    </p>

    <h3><a id="streams_api_changes_210" href="#streams_api_changes_210">Streams API changes in 2.1.0</a></h3>
    <p>
        We updated <code>TopologyDescription</code> API to allow for better runtime checking.
        Users are encouraged to use <code>#topicSet()</code> and <code>#topicPattern()</code> accordingly on <code>TopologyDescription.Source</code> nodes,
        instead of using <code>#topics()</code>, which has since been deprecated. Similarly, use <code>#topic()</code> and <code>#topicNameExtractor()</code>
        to get descriptions of <code>TopologyDescription.Sink</code> nodes. For more details, see
        <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-321%3A+Update+TopologyDescription+to+better+represent+Source+and+Sink+Nodes">KIP-321</a>.
    </p>

    <p>
        We've added a new class <code>Grouped</code> and deprecated <code>Serialized</code>.  The intent of adding <code>Grouped</code> is the ability to
        name repartition topics created when performing aggregation operations.  Users can name the potential repartition topic using the
        <code>Grouped#as()</code> method which takes a <code>String</code> and is used as part of the repartition topic name.  The resulting repartition
        topic name will still follow the pattern of <code>${application-id}-&gt;name&lt;-repartition</code>.  The <code>Grouped</code> class is now favored over
        <code>Serialized</code> in <code>KStream#groupByKey()</code>, <code>KStream#groupBy()</code>, and <code>KTable#groupBy()</code>.
        Note that Kafka Streams does not automatically create repartition topics for aggregation operations.

        Additionally, we've updated the <code>Joined</code> class with a new method <code>Joined#withName</code>
        enabling users to name any repartition topics required for performing Stream/Stream or Stream/Table join.  For more details repartition
        topic naming, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-372%3A+Naming+Repartition+Topics+for+Joins+and+Grouping">KIP-372</a>.

        As a result we've updated the Kafka Streams Scala API and removed the <code>Serialized</code> class in favor of adding <code>Grouped</code>.
        If you just rely on the implicit <code>Serialized</code>, you just need to recompile; if you pass in <code>Serialized</code> explicitly, sorry you'll have to make code changes.
    </p>

    <p>
        We've added a new config named <code>max.task.idle.ms</code> to allow users specify how to handle out-of-order data within a task that may be processing multiple
        topic-partitions (see <a href="/{{version}}/documentation/streams/core-concepts.html#streams_out_of_ordering">Out-of-Order Handling</a> section for more details).
        The default value is set to <code>0</code>, to favor minimized latency over synchronization between multiple input streams from topic-partitions.
        If users would like to wait for longer time when some of the topic-partitions do not have data available to process and hence cannot determine its corresponding stream time,
        they can override this config to a larger value.
    </p>

    <p>
        We've added the missing <code>SessionBytesStoreSupplier#retentionPeriod()</code> to be consistent with the <code>WindowBytesStoreSupplier</code> which allows users to get the specified retention period for session-windowed stores.
        We've also added the missing <code>StoreBuilder#withCachingDisabled()</code> to allow users to turn off caching for their customized stores.
    </p>

    <p>
        We added a new serde for UUIDs (<code>Serdes.UUIDSerde</code>) that you can use via <code>Serdes.UUID()</code>
	(cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-206%3A+Add+support+for+UUID+serialization+and+deserialization">KIP-206</a>).
    </p>

    <p>
        We updated a list of methods that take <code>long</code> arguments as either timestamp (fix point) or duration (time period)
        and replaced them with <code>Instant</code> and <code>Duration</code> parameters for improved semantics.
        Some old methods base on <code>long</code> are deprecated and users are encouraged to update their code.
        <br />
        In particular, aggregation windows (hopping/tumbling/unlimited time windows and session windows) as well as join windows now take <code>Duration</code>
        arguments to specify window size, hop, and gap parameters.
        Also, window sizes and retention times are now specified as <code>Duration</code> type in <code>Stores</code> class.
        The <code>Window</code> class has new methods <code>#startTime()</code> and <code>#endTime()</code> that return window start/end timestamp as <code>Instant</code>.
        For interactive queries, there are new <code>#fetch(...)</code> overloads taking <code>Instant</code> arguments.
        Additionally, punctuations are now registerd via <code>ProcessorContext#schedule(Duration interval, ...)</code>.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times">KIP-358</a>.
    </p>

    <p>
        We deprecated <code>KafkaStreams#close(...)</code> and replaced it with <code>KafkaStreams#close(Duration)</code> that accepts a single timeout argument
        Note: the new <code>#close(Duration)</code> method has improved (but slightly different) semantics.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times">KIP-358</a>.
    </p>

    <p>
        The newly exposed <code>AdminClient</code> metrics are now available when calling the <code>KafkaStream#metrics()</code> method.
        For more details on exposing <code>AdminClients</code> metrics
        see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-324%3A+Add+method+to+get+metrics%28%29+in+AdminClient">KIP-324</a>
    </p>

    <p>
        We deprecated the notion of segments in window stores as those are intended to be an implementation details.
        Thus, method <code>Windows#segments()</code> and variable <code>Windows#segments</code> were deprecated.
        If you implement custom windows, you should update your code accordingly.
        Similarly, <code>WindowBytesStoreSupplier#segments()</code> was deprecated and replaced with <code>WindowBytesStoreSupplier#segmentInterval()</code>.
        If you implement custom window store, you need to update your code accordingly.
	    Finally, <code>Stores#persistentWindowStore(...)</code> were deprecated and replaced with a new overload that does not allow to specify the number of segments any longer.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-319%3A+Replace+segments+with+segmentInterval+in+WindowBytesStoreSupplier">KIP-319</a>
        (note: <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-328%3A+Ability+to+suppress+updates+for+KTables">KIP-328</a> and 
	    <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times">KIP-358</a> 'overlap' with KIP-319).
    </p>

    <p>
        We've added an overloaded <code>StreamsBuilder#build</code> method that accepts an instance of <code>java.util.Properties</code> with the intent of using the
        <code>StreamsConfig#TOPOLOGY_OPTIMIZATION</code> config added in Kafka Streams 2.0. Before 2.1, when building a topology with
        the DSL, Kafka Streams writes the physical plan as the user makes calls on the DSL.  Now by providing a <code>java.util.Properties</code> instance when
        executing a <code>StreamsBuilder#build</code> call, Kafka Streams can optimize the physical plan of the topology, provided the <code>StreamsConfig#TOPOLOGY_OPTIMIZATION</code>
        config is set to <code>StreamsConfig#OPTIMIZE</code>.  By setting <code>StreamsConfig#OPTIMIZE</code> in addition to the <code>KTable</code> optimization of
        reusing the source topic as the changelog topic, the topology may be optimized to merge redundant repartition topics into one
        repartition topic.  The original no parameter version of <code>StreamsBuilder#build</code> is still available for those who wish to not
        optimize their topology.  Note that enabling optimization of the topology may require you to do an application reset when redeploying the application.  For more
        details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-312%3A+Add+Overloaded+StreamsBuilder+Build+Method+to+Accept+java.util.Properties">KIP-312</a>
    </p>

    <h3><a id="streams_api_changes_200" href="#streams_api_changes_200">Streams API changes in 2.0.0</a></h3>
    <p>
        In 2.0.0 we have added a few new APIs on the <code>ReadOnlyWindowStore</code> interface (for details please read <a href="#streams_api_changes_200">Streams API changes</a> below).
        If you have customized window store implementations that extends the <code>ReadOnlyWindowStore</code> interface you need to make code changes.
    </p>

    <p>
        In addition, if you using Java 8 method references in your Kafka Streams code you might need to update your code to resolve method ambiguities.
        Hot-swapping the jar-file only might not work for this case.
        See below a complete list of <a href="#streams_api_changes_200">2.0.0</a>
        API and semantic changes that allow you to advance your application and/or simplify your code base.
    </p>

    <p>
        We moved <code>Consumed</code> interface from <code>org.apache.kafka.streams</code> to <code>org.apache.kafka.streams.kstream</code>
        as it was mistakenly placed in the previous release. If your code has already used it there is a simple one-liner change needed in your import statement.
    </p>

    <p>
        We have also removed some public APIs that are deprecated prior to 1.0.x in 2.0.0.
        See below for a detailed list of removed APIs.
    </p>
    <p>
        We have removed the <code>skippedDueToDeserializationError-rate</code> and <code>skippedDueToDeserializationError-total</code> metrics.
        Deserialization errors, and all other causes of record skipping, are now accounted for in the pre-existing metrics
        <code>skipped-records-rate</code> and <code>skipped-records-total</code>. When a record is skipped, the event is
        now logged at WARN level. If these warnings become burdensome, we recommend explicitly filtering out unprocessable
        records instead of depending on record skipping semantics. For more details, see
        <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-274%3A+Kafka+Streams+Skipped+Records+Metrics">KIP-274</a>.
        As of right now, the potential causes of skipped records are:
    </p>
    <ul>
        <li><code>null</code> keys in table sources</li>
        <li><code>null</code> keys in table-table inner/left/outer/right joins</li>
        <li><code>null</code> keys or values in stream-table joins</li>
        <li><code>null</code> keys or values in stream-stream joins</li>
        <li><code>null</code> keys or values in aggregations on grouped streams</li>
        <li><code>null</code> keys or values in reductions on grouped streams</li>
        <li><code>null</code> keys in aggregations on windowed streams</li>
        <li><code>null</code> keys in reductions on windowed streams</li>
        <li><code>null</code> keys in aggregations on session-windowed streams</li>
        <li>
            Errors producing results, when the configured <code>default.production.exception.handler</code> decides to
            <code>CONTINUE</code> (the default is to <code>FAIL</code> and throw an exception).
        </li>
        <li>
            Errors deserializing records, when the configured <code>default.deserialization.exception.handler</code>
            decides to <code>CONTINUE</code> (the default is to <code>FAIL</code> and throw an exception).
            This was the case previously captured in the <code>skippedDueToDeserializationError</code> metrics.
        </li>
        <li>Fetched records having a negative timestamp.</li>
    </ul>

    <p>
        We've also fixed the metrics name for time and session windowed store operations in 2.0. As a result, our current built-in stores
        will have their store types in the metric names as <code>in-memory-state</code>, <code>in-memory-lru-state</code>,
        <code>rocksdb-state</code>, <code>rocksdb-window-state</code>, and <code>rocksdb-session-state</code>. For example, a RocksDB time windowed store's
        put operation metrics would now be
        <code>kafka.streams:type=stream-rocksdb-window-state-metrics,client-id=([-.\w]+),task-id=([-.\w]+),rocksdb-window-state-id=([-.\w]+)</code>.
        Users need to update their metrics collecting and reporting systems for their time and session windowed stores accordingly.
        For more details, please read the <a href="/{{version}}/documentation/#kafka_streams_store_monitoring">State Store Metrics</a> section.
    </p>

    <p>
        We have added support for methods in <code>ReadOnlyWindowStore</code> which allows for querying a single window's key-value pair.
        For users who have customized window store implementations on the above interface, they'd need to update their code to implement the newly added method as well.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-261%3A+Add+Single+Value+Fetch+in+Window+Stores">KIP-261</a>.
    </p>
    <p>
        We have added public <code>WindowedSerdes</code> to allow users to read from / write to a topic storing windowed table changelogs directly.
        In addition, in <code>StreamsConfig</code> we have also added <code>default.windowed.key.serde.inner</code> and <code>default.windowed.value.serde.inner</code>
        to let users specify inner serdes if the default serde classes are windowed serdes.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-265%3A+Make+Windowed+Serde+to+public+APIs">KIP-265</a>.
    </p>
    <p>
        We've added message header support in the <code>Processor API</code> in Kafka 2.0.0. In particular, we have added a new API <code>ProcessorContext#headers()</code>
        which returns a <code>Headers</code> object that keeps track of the headers of the source topic's message that is being processed. Through this object, users can manipulate
        the headers map that is being propagated throughout the processor topology as well. For more details please feel free to read
        the <a href="/{{version}}/documentation/streams/developer-guide/processor-api.html#accessing-processor-context">Developer Guide</a> section.
    </p>
    <p>
        We have deprecated constructors of <code>KafkaStreams</code> that take a <code>StreamsConfig</code> as parameter.
        Please use the other corresponding constructors that accept <code>java.util.Properties</code> instead.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-245%3A+Use+Properties+instead+of+StreamsConfig+in+KafkaStreams+constructor">KIP-245</a>.
    </p>
    <p>
        Kafka 2.0.0 allows to manipulate timestamps of output records using the Processor API (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-251%3A+Allow+timestamp+manipulation+in+Processor+API">KIP-251</a>).
        To enable this new feature, <code>ProcessorContext#forward(...)</code> was modified.
        The two existing overloads <code>#forward(Object key, Object value, String childName)</code> and <code>#forward(Object key, Object value, int childIndex)</code> were deprecated and a new overload <code>#forward(Object key, Object value, To to)</code> was added.
        The new class <code>To</code> allows you to send records to all or specific downstream processors by name and to set the timestamp for the output record.
        Forwarding based on child index is not supported in the new API any longer.
    </p>
    <p>
        We have added support to allow routing records dynamically to Kafka topics. More specifically, in both the lower-level <code>Topology#addSink</code> and higher-level <code>KStream#to</code> APIs, we have added variants that
        take a <code>TopicNameExtractor</code> instance instead of a specific <code>String</code> typed topic name, such that for each received record from the upstream processor, the library will dynamically determine which Kafka topic to write to
        based on the record's key and value, as well as record context. Note that all the Kafka topics that may possibly be used are still considered as user topics and hence required to be pre-created. In addition to that, we have modified the
        <code>StreamPartitioner</code> interface to add the topic name parameter since the topic name now may not be known beforehand; users who have customized implementations of this interface would need to update their code while upgrading their application
        to use Kafka Streams 2.0.0.
    </p>
    <p>
        <a href="https://cwiki.apache.org/confluence/x/DVyHB">KIP-284</a> changed the retention time for repartition topics by setting its default value to <code>Long.MAX_VALUE</code>.
        Instead of relying on data retention Kafka Streams uses the new purge data API to delete consumed data from those topics and to keep used storage small now.
    </p>
    <p>
        We have modified the <code>ProcessorStateManger#register(...)</code> signature and removed the deprecated <code>loggingEnabled</code> boolean parameter as it is specified in the <code>StoreBuilder</code>.
        Users who used this function to register their state stores into the processor topology need to simply update their code and remove this parameter from the caller.
    </p>
    <p>
        Kafka Streams DSL for Scala is a new Kafka Streams client library available for developers authoring Kafka Streams applications in Scala.  It wraps core Kafka Streams DSL types to make it easier to call when
        interoperating with Scala code.  For example, it includes higher order functions as parameters for transformations avoiding the need anonymous classes in Java 7 or experimental SAM type conversions in Scala 2.11,
        automatic conversion between Java and Scala collection types, a way
        to implicitly provide SerDes to reduce boilerplate from your application and make it more typesafe, and more!  For more information see the
        <a href="/{{version}}/documentation/streams/developer-guide/dsl-api.html#scala-dsl">Kafka Streams DSL for Scala documentation</a> and
        <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-270+-+A+Scala+Wrapper+Library+for+Kafka+Streams">KIP-270</a>.
    </p>
    <p>
        We have removed these deprecated APIs:
    </p>
    <ul>
        <li><code>KafkaStreams#toString</code> no longer returns the topology and runtime metadata; to get topology metadata users can call <code>Topology#describe()</code> and to get thread runtime metadata users can call <code>KafkaStreams#localThreadsMetadata</code> (they are deprecated since 1.0.0).
            For detailed guidance on how to update your code please read <a href="#streams_api_changes_100">here</a></li>
        <li><code>TopologyBuilder</code> and <code>KStreamBuilder</code> are removed and replaced by <code>Topology</code> and <code>StreamsBuidler</code> respectively (they are deprecated since 1.0.0).
            For detailed guidance on how to update your code please read <a href="#streams_api_changes_100">here</a></li>
        <li><code>StateStoreSupplier</code> are removed and replaced with <code>StoreBuilder</code> (they are deprecated since 1.0.0);
            and the corresponding <code>Stores#create</code> and <code>KStream, KTable, KGroupedStream</code> overloaded functions that use it have also been removed.
            For detailed guidance on how to update your code please read <a href="#streams_api_changes_100">here</a></li>
        <li><code>KStream, KTable, KGroupedStream</code> overloaded functions that requires serde and other specifications explicitly are removed and replaced with simpler overloaded functions that use <code>Consumed, Produced, Serialized, Materialized, Joined</code> (they are deprecated since 1.0.0).
            For detailed guidance on how to update your code please read <a href="#streams_api_changes_100">here</a></li>
        <li><code>Processor#punctuate</code>, <code>ValueTransformer#punctuate</code>, <code>ValueTransformer#punctuate</code> and <code>ProcessorContext#schedule(long)</code> are removed and replaced by <code>ProcessorContext#schedule(long, PunctuationType, Punctuator)</code> (they are deprecated in 1.0.0). </li>
        <li>The second <code>boolean</code> typed parameter "loggingEnabled" in <code>ProcessorContext#register</code> has been removed; users can now use <code>StoreBuilder#withLoggingEnabled, withLoggingDisabled</code> to specify the behavior when they create the state store. </li>
        <li><code>KTable#writeAs, print, foreach, to, through</code> are removed, users can call <code>KTable#tostream()#writeAs</code> instead for the same purpose (they are deprecated since 0.11.0.0).
            For detailed list of removed APIs please read <a href="#streams_api_changes_0110">here</a></li>
        <li><code>StreamsConfig#KEY_SERDE_CLASS_CONFIG, VALUE_SERDE_CLASS_CONFIG, TIMESTAMP_EXTRACTOR_CLASS_CONFIG</code> are removed and replaced with <code>StreamsConfig#DEFAULT_KEY_SERDE_CLASS_CONFIG, DEFAULT_VALUE_SERDE_CLASS_CONFIG, DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG</code> respectively (they are deprecated since 0.11.0.0). </li>
        <li><code>StreamsConfig#ZOOKEEPER_CONNECT_CONFIG</code> are removed as we do not need ZooKeeper dependency in Streams any more (it is deprecated since 0.10.2.0). </li>
    </ul>

    <h3><a id="streams_api_changes_110" href="#streams_api_changes_110">Streams API changes in 1.1.0</a></h3>
    <p>
        We have added support for methods in <code>ReadOnlyWindowStore</code> which allows for querying <code>WindowStore</code>s without the necessity of providing keys.
        For users who have customized window store implementations on the above interface, they'd need to update their code to implement the newly added method as well.
        For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-205%3A+Add+all%28%29+and+range%28%29+API+to+ReadOnlyWindowStore">KIP-205</a>.
    </p>

    <p>
	    There is a new artifact <code>kafka-streams-test-utils</code> providing a <code>TopologyTestDriver</code>, <code>ConsumerRecordFactory</code>, and <code>OutputVerifier</code> class.
	    You can include the new artifact as a regular dependency to your unit tests and use the test driver to test your business logic of your Kafka Streams application.
	    For more details, see <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-247%3A+Add+public+test+utils+for+Kafka+Streams">KIP-247</a>.
    </p>

    <p>
        The introduction of <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-220%3A+Add+AdminClient+into+Kafka+Streams%27+ClientSupplier">KIP-220</a>
        enables you to provide configuration parameters for the embedded admin client created by Kafka Streams, similar to the embedded producer and consumer clients.
        You can provide the configs via <code>StreamsConfig</code> by adding the configs with the prefix <code>admin.</code> as defined by <code>StreamsConfig#adminClientPrefix(String)</code>
        to distinguish them from configurations of other clients that share the same config names.
    </p>

    <p>
        New method in <code>KTable</code>
    </p>
    <ul>
        <li> <code>transformValues</code> methods have been added to <code>KTable</code>. Similar to those on <code>KStream</code>, these methods allow for richer, stateful, value transformation similar to the Processor API.</li>
    </ul>

    <p>
	New method in <code>GlobalKTable</code>
    </p>
    <ul>
	<li> A method has been provided such that it will return the store name associated with the <code>GlobalKTable</code> or <code>null</code> if the store name is non-queryable. </li>
    </ul>

    <p>
        New methods in <code>KafkaStreams</code>:
    </p>
    <ul>
        <li> added overload for the constructor that allows overriding the <code>Time</code> object used for tracking system wall-clock time; this is useful for unit testing your application code. </li>
    </ul>

    <p> New methods in <code>KafkaClientSupplier</code>: </p>
    <ul>
        <li> added <code>getAdminClient(config)</code> that allows to override an <code>AdminClient</code> used for administrative requests such as internal topic creations, etc. </li>
    </ul>

    <p>New error handling for exceptions during production:</p>
    <ul>
        <li>added interface <code>ProductionExceptionHandler</code> that allows implementors to decide whether or not Streams should <code>FAIL</code> or <code>CONTINUE</code> when certain exception occur while trying to produce.</li>
        <li>provided an implementation, <code>DefaultProductionExceptionHandler</code> that always fails, preserving the existing behavior by default.</li>
        <li>changing which implementation is used can be done by settings <code>default.production.exception.handler</code> to the fully qualified name of a class implementing this interface.</li>
    </ul>

    <p> Changes in <code>StreamsResetter</code>: </p>
    <ul>
        <li> added options to specify input topics offsets to reset according to <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-171+-+Extend+Consumer+Group+Reset+Offset+for+Stream+Application">KIP-171</a></li>
    </ul>

    <h3><a id="streams_api_changes_100" href="#streams_api_changes_100">Streams API changes in 1.0.0</a></h3>

    <p>
        With 1.0 a major API refactoring was accomplished and the new API is cleaner and easier to use.
        This change includes the five main classes <code>KafkaStreams</code>, <code>KStreamBuilder</code>,
        <code>KStream</code>, <code>KTable</code>, and <code>TopologyBuilder</code> (and some more others).
        All changes are fully backward compatible as old API is only deprecated but not removed.
        We recommend to move to the new API as soon as you can.
        We will summarize all API changes in the next paragraphs.
    </p>

    <p>
        The two main classes to specify a topology via the DSL (<code>KStreamBuilder</code>)
        or the Processor API (<code>TopologyBuilder</code>) were deprecated and replaced by
        <code>StreamsBuilder</code> and <code>Topology</code> (both new classes are located in
        package <code>org.apache.kafka.streams</code>).
        Note, that <code>StreamsBuilder</code> does not extend <code>Topology</code>, i.e.,
        the class hierarchy is different now.
        The new classes have basically the same methods as the old ones to build a topology via DSL or Processor API.
        However, some internal methods that were public in <code>KStreamBuilder</code>
        and <code>TopologyBuilder</code> but not part of the actual API are not present
        in the new classes any longer.
        Furthermore, some overloads were simplified compared to the original classes.
        See <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-120%3A+Cleanup+Kafka+Streams+builder+API">KIP-120</a>
        and <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-182%3A+Reduce+Streams+DSL+overloads+and+allow+easier+use+of+custom+storage+engines">KIP-182</a>
        for full details.
    </p>

    <p>
        Changing how a topology is specified also affects <code>KafkaStreams</code> constructors,
        that now only accept a <code>Topology</code>.
        Using the DSL builder class <code>StreamsBuilder</code> one can get the constructed
        <code>Topology</code> via <code>StreamsBuilder#build()</code>.
        Additionally, a new class <code>org.apache.kafka.streams.TopologyDescription</code>
        (and some more dependent classes) were added.
        Those can be used to get a detailed description of the specified topology
        and can be obtained by calling <code>Topology#describe()</code>.
        An example using this new API is shown in the <a href="/{{version}}/documentation/streams/quickstart">quickstart section</a>.
    </p>

    <p>
        New methods in <code>KStream</code>:
    </p>
    <ul>
        <li>With the introduction of <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-202+Move+merge%28%29+from+StreamsBuilder+to+KStream">KIP-202</a>
            a new method <code>merge()</code> has been created in <code>KStream</code> as the StreamsBuilder class's <code>StreamsBuilder#merge()</code> has been removed.
            The method signature was also changed, too: instead of providing multiple <code>KStream</code>s into the method at the once, only a single <code>KStream</code> is accepted.
        </li>
    </ul>

    <p>
        New methods in <code>KafkaStreams</code>:
    </p>
    <ul>
        <li>retrieve the current runtime information about the local threads via <code>localThreadsMetadata()</code> </li>
        <li>observe the restoration of all state stores via <code>setGlobalStateRestoreListener()</code>, in which users can provide their customized implementation of the <code>org.apache.kafka.streams.processor.StateRestoreListener</code> interface</li>
    </ul>

    <p>
        Deprecated / modified methods in <code>KafkaStreams</code>:
    </p>
    <ul>
        <li>
            <code>toString()</code>, <code>toString(final String indent)</code> were previously used to return static and runtime information.
            They have been deprecated in favor of using the new classes/methods <code>localThreadsMetadata()</code> / <code>ThreadMetadata</code> (returning runtime information) and
            <code>TopologyDescription</code> / <code>Topology#describe()</code> (returning static information).
        </li>
        <li>
            With the introduction of <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-182%3A+Reduce+Streams+DSL+overloads+and+allow+easier+use+of+custom+storage+engines">KIP-182</a>
            you should no longer pass in <code>Serde</code> to <code>KStream#print</code> operations.
            If you can't rely on using <code>toString</code> to print your keys an values, you should instead you provide a custom <code>KeyValueMapper</code> via the <code>Printed#withKeyValueMapper</code> call.
        </li>
        <li>
            <code>setStateListener()</code> now can only be set before the application start running, i.e. before <code>KafkaStreams.start()</code> is called.
        </li>
    </ul>

    <p>
        Deprecated methods in <code>KGroupedStream</code>
    </p>
    <ul>
        <li>
            Windowed aggregations have been deprecated from <code>KGroupedStream</code> and moved to <code>WindowedKStream</code>.
            You can now perform a windowed aggregation by, for example, using <code>KGroupedStream#windowedBy(Windows)#reduce(Reducer)</code>.
        </li>
    </ul>

    <p>
        Modified methods in <code>Processor</code>:
    </p>
    <ul>
        <li>
            <p>
                The Processor API was extended to allow users to schedule <code>punctuate</code> functions either based on data-driven <b>stream time</b> or wall-clock time.
                As a result, the original <code>ProcessorContext#schedule</code> is deprecated with a new overloaded function that accepts a user customizable <code>Punctuator</code> callback interface, which triggers its <code>punctuate</code> API method periodically based on the <code>PunctuationType</code>.
                The <code>PunctuationType</code> determines what notion of time is used for the punctuation scheduling: either <a href="/{{version}}/documentation/streams/core-concepts#streams_time">stream time</a> or wall-clock time (by default, <b>stream time</b> is configured to represent event time via <code>TimestampExtractor</code>).
                In addition, the <code>punctuate</code> function inside <code>Processor</code> is also deprecated.
            </p>
            <p>
                Before this, users could only schedule based on stream time (i.e. <code>PunctuationType.STREAM_TIME</code>) and hence the <code>punctuate</code> function was data-driven only because stream time is determined (and advanced forward) by the timestamps derived from the input data.
                If there is no data arriving at the processor, the stream time would not advance and hence punctuation will not be triggered.
                On the other hand, When wall-clock time (i.e. <code>PunctuationType.WALL_CLOCK_TIME</code>) is used, <code>punctuate</code> will be triggered purely based on wall-clock time.
                So for example if the <code>Punctuator</code> function is scheduled based on <code>PunctuationType.WALL_CLOCK_TIME</code>, if these 60 records were processed within 20 seconds,
                <code>punctuate</code> would be called 2 times (one time every 10 seconds);
                if these 60 records were processed within 5 seconds, then no <code>punctuate</code> would be called at all.
                Users can schedule multiple <code>Punctuator</code> callbacks with different <code>PunctuationType</code>s within the same processor by simply calling <code>ProcessorContext#schedule</code> multiple times inside processor's <code>init()</code> method.
            </p>
        </li>
    </ul>

    <p>
        If you are monitoring on task level or processor-node / state store level Streams metrics, please note that the metrics sensor name and hierarchy was changed:
        The task ids, store names and processor names are no longer in the sensor metrics names, but instead are added as tags of the sensors to achieve consistent metrics hierarchy.
        As a result you may need to make corresponding code changes on your metrics reporting and monitoring tools when upgrading to 1.0.0.
        Detailed metrics sensor can be found in the <a href="#kafka_streams_monitoring">Streams Monitoring</a> section.
    </p>

    <p>
        The introduction of <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+deserialization+exception+handlers">KIP-161</a>
        enables you to provide a default exception handler for deserialization errors when reading data from Kafka rather than throwing the exception all the way out of your streams application.
        You can provide the configs via the <code>StreamsConfig</code> as <code>StreamsConfig#DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG</code>.
        The specified handler must implement the <code>org.apache.kafka.streams.errors.DeserializationExceptionHandler</code> interface.
    </p>

    <p>
        The introduction of <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-173%3A+Add+prefix+to+StreamsConfig+to+enable+setting+default+internal+topic+configs">KIP-173</a>
        enables you to provide topic configuration parameters for any topics created by Kafka Streams.
        This includes repartition and changelog topics.
        You can provide the configs via the <code>StreamsConfig</code> by adding the configs with the prefix as defined by <code>StreamsConfig#topicPrefix(String)</code>.
        Any properties in the <code>StreamsConfig</code> with the prefix will be applied when creating internal topics.
        Any configs that aren't topic configs will be ignored.
        If you already use <code>StateStoreSupplier</code> or <code>Materialized</code> to provide configs for changelogs, then they will take precedence over those supplied in the config.
    </p>

    <h3><a id="streams_api_changes_0110" href="#streams_api_changes_0110">Streams API changes in 0.11.0.0</a></h3>

    <p> Updates in <code>StreamsConfig</code>: </p>
    <ul>
        <li> new configuration parameter <code>processing.guarantee</code> is added </li>
        <li> configuration parameter <code>key.serde</code> was deprecated and replaced by <code>default.key.serde</code> </li>
        <li> configuration parameter <code>value.serde</code> was deprecated and replaced by <code>default.value.serde</code> </li>
        <li> configuration parameter <code>timestamp.extractor</code> was deprecated and replaced by <code>default.timestamp.extractor</code> </li>
        <li> method <code>keySerde()</code> was deprecated and replaced by <code>defaultKeySerde()</code> </li>
        <li> method <code>valueSerde()</code> was deprecated and replaced by <code>defaultValueSerde()</code> </li>
        <li> new method <code>defaultTimestampExtractor()</code> was added </li>
    </ul>

    <p> New methods in <code>TopologyBuilder</code>: </p>
    <ul>
        <li> added overloads for <code>addSource()</code> that allow to define a <code>TimestampExtractor</code> per source node </li>
        <li> added overloads for <code>addGlobalStore()</code> that allow to define a <code>TimestampExtractor</code> per source node associated with the global store </li>
    </ul>

    <p> New methods in <code>KStreamBuilder</code>: </p>
    <ul>
        <li> added overloads for <code>stream()</code> that allow to define a <code>TimestampExtractor</code> per input stream </li>
        <li> added overloads for <code>table()</code> that allow to define a <code>TimestampExtractor</code> per input table </li>
        <li> added overloads for <code>globalKTable()</code> that allow to define a <code>TimestampExtractor</code> per global table </li>
    </ul>

    <p> Deprecated methods in <code>KTable</code>: </p>
    <ul>
        <li> <code>void foreach(final ForeachAction&lt;? super K, ? super V&gt; action)</code> </li>
        <li> <code>void print()</code> </li>
        <li> <code>void print(final String streamName)</code> </li>
        <li> <code>void print(final Serde&lt;K&gt; keySerde, final Serde&lt;V&gt; valSerde)</code> </li>
        <li> <code>void print(final Serde&lt;K&gt; keySerde, final Serde&lt;V&gt; valSerde, final String streamName)</code> </li>
        <li> <code>void writeAsText(final String filePath)</code> </li>
        <li> <code>void writeAsText(final String filePath, final String streamName)</code> </li>
        <li> <code>void writeAsText(final String filePath, final Serde&lt;K&gt; keySerde, final Serde&lt;V&gt; valSerde)</code> </li>
        <li> <code>void writeAsText(final String filePath, final String streamName, final Serde&lt;K&gt; keySerde, final Serde&lt;V&gt; valSerde)</code> </li>
    </ul>

    <p>
        The above methods have been deprecated in favor of using the Interactive Queries API.
        If you want to query the current content of the state store backing the KTable, use the following approach:
    </p>
    <ul>
        <li> Make a call to <code>KafkaStreams.store(final String storeName, final QueryableStoreType&lt;T&gt; queryableStoreType)</code> </li>
        <li> Then make a call to <code>ReadOnlyKeyValueStore.all()</code> to iterate over the keys of a <code>KTable</code>. </li>
    </ul>
    <p>
        If you want to view the changelog stream of the <code>KTable</code> then you could call <code>KTable.toStream().print(Printed.toSysOut)</code>.
    </p>

    <p> Metrics using exactly-once semantics: </p>
    <p>
        If exactly-once processing is enabled via the <code>processing.guarantees</code> parameter, internally Streams switches from a producer per thread to a producer per task runtime model.
        In order to distinguish the different producers, the producer's <code>client.id</code> additionally encodes the task-ID for this case.
        Because the producer's <code>client.id</code> is used to report JMX metrics, it might be required to update tools that receive those metrics.

    </p>

    <p> Producer's <code>client.id</code> naming schema: </p>
    <ul>
        <li> at-least-once (default): <code>[client.Id]-StreamThread-[sequence-number]</code> </li>
        <li> exactly-once: <code>[client.Id]-StreamThread-[sequence-number]-[taskId]</code> </li>
    </ul>
    <p> <code>[client.Id]</code> is either set via Streams configuration parameter <code>client.id</code> or defaults to <code>[application.id]-[processId]</code> (<code>[processId]</code> is a random UUID). </p>

    <h3><a id="streams_api_changes_01021" href="#streams_api_changes_01021">Notable changes in 0.10.2.1</a></h3>

    <p>
        Parameter updates in <code>StreamsConfig</code>:
    </p>
    <ul>
        <li> The default config values of embedded producer's <code>retries</code> and consumer's <code>max.poll.interval.ms</code> have been changed to improve the resiliency of a Kafka Streams application </li>
    </ul>

    <h3><a id="streams_api_changes_0102" href="#streams_api_changes_0102">Streams API changes in 0.10.2.0</a></h3>

    <p>
        New methods in <code>KafkaStreams</code>:
    </p>
    <ul>
        <li> set a listener to react on application state change via <code>setStateListener(StateListener listener)</code> </li>
        <li> retrieve the current application state via <code>state()</code> </li>
        <li> retrieve the global metrics registry via <code>metrics()</code> </li>
        <li> apply a timeout when closing an application via <code>close(long timeout, TimeUnit timeUnit)</code> </li>
        <li> specify a custom indent when retrieving Kafka Streams information via <code>toString(String indent)</code> </li>
    </ul>

    <p>
        Parameter updates in <code>StreamsConfig</code>:
    </p>
    <ul>
        <li> parameter <code>zookeeper.connect</code> was deprecated; a Kafka Streams application does no longer interact with ZooKeeper for topic management but uses the new broker admin protocol
            (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-4+-+Command+line+and+centralized+administrative+operations#KIP-4-Commandlineandcentralizedadministrativeoperations-TopicAdminSchema.1">KIP-4, Section "Topic Admin Schema"</a>) </li>
        <li> added many new parameters for metrics, security, and client configurations </li>
    </ul>

    <p> Changes in <code>StreamsMetrics</code> interface: </p>
    <ul>
        <li> removed methods: <code>addLatencySensor()</code> </li>
        <li> added methods: <code>addLatencyAndThroughputSensor()</code>, <code>addThroughputSensor()</code>, <code>recordThroughput()</code>,
            <code>addSensor()</code>, <code>removeSensor()</code> </li>
    </ul>

    <p> New methods in <code>TopologyBuilder</code>: </p>
    <ul>
        <li> added overloads for <code>addSource()</code> that allow to define a <code>auto.offset.reset</code> policy per source node </li>
        <li> added methods <code>addGlobalStore()</code> to add global <code>StateStore</code>s </li>
    </ul>

    <p> New methods in <code>KStreamBuilder</code>: </p>
    <ul>
        <li> added overloads for <code>stream()</code> and <code>table()</code> that allow to define a <code>auto.offset.reset</code> policy per input stream/table </li>
        <li> added method <code>globalKTable()</code> to create a <code>GlobalKTable</code> </li>
    </ul>

    <p> New joins for <code>KStream</code>: </p>
    <ul>
        <li> added overloads for <code>join()</code> to join with <code>KTable</code> </li>
        <li> added overloads for <code>join()</code> and <code>leftJoin()</code> to join with <code>GlobalKTable</code> </li>
        <li> note, join semantics in 0.10.2 were improved and thus you might see different result compared to 0.10.0.x and 0.10.1.x
            (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Join+Semantics">Kafka Streams Join Semantics</a> in the Apache Kafka wiki)
    </ul>

    <p> Aligned <code>null</code>-key handling for <code>KTable</code> joins: </p>
    <ul>
        <li> like all other KTable operations, <code>KTable-KTable</code> joins do not throw an exception on <code>null</code> key records anymore, but drop those records silently </li>
    </ul>

    <p> New window type <em>Session Windows</em>: </p>
    <ul>
        <li> added class <code>SessionWindows</code> to specify session windows </li>
        <li> added overloads for <code>KGroupedStream</code> methods <code>count()</code>, <code>reduce()</code>, and <code>aggregate()</code>
            to allow session window aggregations </li>
    </ul>

    <p> Changes to <code>TimestampExtractor</code>: </p>
    <ul>
        <li> method <code>extract()</code> has a second parameter now </li>
        <li> new default timestamp extractor class <code>FailOnInvalidTimestamp</code>
            (it gives the same behavior as old (and removed) default extractor <code>ConsumerRecordTimestampExtractor</code>) </li>
        <li> new alternative timestamp extractor classes <code>LogAndSkipOnInvalidTimestamp</code> and <code>UsePreviousTimeOnInvalidTimestamps</code> </li>
    </ul>

    <p> Relaxed type constraints of many DSL interfaces, classes, and methods (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-100+-+Relax+Type+constraints+in+Kafka+Streams+API">KIP-100</a>). </p>

    <h3><a id="streams_api_changes_0101" href="#streams_api_changes_0101">Streams API changes in 0.10.1.0</a></h3>

    <p> Stream grouping and aggregation split into two methods: </p>
    <ul>
        <li> old: KStream #aggregateByKey(), #reduceByKey(), and #countByKey() </li>
        <li> new: KStream#groupByKey() plus KGroupedStream #aggregate(), #reduce(), and #count() </li>
        <li> Example: stream.countByKey() changes to stream.groupByKey().count() </li>
    </ul>

    <p> Auto Repartitioning: </p>
    <ul>
        <li> a call to through() after a key-changing operator and before an aggregation/join is no longer required </li>
        <li> Example: stream.selectKey(...).through(...).countByKey() changes to stream.selectKey().groupByKey().count() </li>
    </ul>

    <p> TopologyBuilder: </p>
    <ul>
        <li> methods #sourceTopics(String applicationId) and #topicGroups(String applicationId) got simplified to #sourceTopics() and #topicGroups() </li>
    </ul>

    <p> DSL: new parameter to specify state store names: </p>
    <ul>
        <li> The new Interactive Queries feature requires to specify a store name for all source KTables and window aggregation result KTables (previous parameter "operator/window name" is now the storeName) </li>
        <li> KStreamBuilder#table(String topic) changes to #topic(String topic, String storeName) </li>
        <li> KTable#through(String topic) changes to #through(String topic, String storeName) </li>
        <li> KGroupedStream #aggregate(), #reduce(), and #count() require additional parameter "String storeName"</li>
        <li> Example: stream.countByKey(TimeWindows.of("windowName", 1000)) changes to stream.groupByKey().count(TimeWindows.of(1000), "countStoreName") </li>
    </ul>

    <p> Windowing: </p>
    <ul>
        <li> Windows are not named anymore: TimeWindows.of("name", 1000) changes to TimeWindows.of(1000) (cf. DSL: new parameter to specify state store names) </li>
        <li> JoinWindows has no default size anymore: JoinWindows.of("name").within(1000) changes to JoinWindows.of(1000) </li>
    </ul>

    <div class="pagination">
        <a href="/{{version}}/documentation/streams/developer-guide/app-reset-tool" class="pagination__btn pagination__btn__prev">Previous</a>
        <a href="#" class="pagination__btn pagination__btn__next pagination__btn--disabled">Next</a>
    </div>
</script>

<!--#include virtual="../../includes/_header.htm" -->
<!--#include virtual="../../includes/_top.htm" -->
<div class="content documentation documentation--current">
    <!--#include virtual="../../includes/_nav.htm" -->
    <div class="right">
        <!--#include virtual="../../includes/_docs_banner.htm" -->
        <ul class="breadcrumbs">
            <li><a href="/documentation">Documentation</a></li>
            <li><a href="/documentation/streams">Kafka Streams</a></li>
        </ul>
        <div class="p-content"></div>
    </div>
</div>
<!--#include virtual="../../includes/_footer.htm" -->
<script>
$(function() {
  // Show selected style on nav item
  $('.b-nav__streams').addClass('selected');

  // Display docs subnav items
  $('.b-nav__docs').parent().toggleClass('nav__item__with__subs--expanded');
});
</script>