0110/streams/upgrade-guide.html - kafka-site - Git at Google

 <!--
  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 &amp; API Changes</h1>

     <p>
         If you want to upgrade from 0.10.2.x to 0.11.0 you don't need to do any code changes as the public API is fully backward compatible.
         However, some configuration parameters were deprecated and thus it is recommend to update your code eventually to allow for future upgrades.
         See <a href="#streams_api_changes_0110">below</a> a complete list of 0.11.0 API and semantical changes that allow you to advance your application and/or simplify your code base, including the usage of new features.
     </p>

     <p>
         If you want to upgrade from 0.10.1.x to 0.11.0, also see the <a href="/{{version}}/upgrade/#upgrade_1020_streams"><b>Upgrade Section for 0.10.2</b></a>.
         It highlights incompatible changes you need to consider to upgrade your code and application.
         See below a complete list of <a href="#streams_api_changes_0102">0.10.2</a> and <a href="#streams_api_changes_0110">0.11.0</a> API and semantical changes
         that allow you to advance your application and/or simplify your code base, including the usage of new features.
     </p>

     <p>
         Upgrading from 0.10.0.x to 0.11.0.x directly is also possible.
         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.
         See <a href="#streams_api_changes_0101">Streams API changes in 0.10.1</a>, <a href="#streams_api_changes_0102">Streams API changes in 0.10.2</a>,
         and <a href="#streams_api_changes_0110">Streams API changes in 0.11.0</a> for a complete list of API changes.
         Upgrading to 0.11.0.3 requires two rolling bounces with config <code>upgrade.from="0.10.0"</code> set for first upgrade phase
         (cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>).
         As an alternative, an offline upgrade is also possible.
     </p>
     <ul>
         <li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to <code>"0.10.0"</code> for new version 0.11.0.3 </li>
         <li> bounce each instance of your application once </li>
         <li> prepare your newly deployed 0.11.0.3 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> Upgrading from 0.10.0.x to 0.11.0.0, 0.11.0.1, or 0.11.0.2 requires an offline upgrade (rolling bounce upgrade is not supported) </p>
     <ul>
         <li> stop all old (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 (0.11.0.0, 0.11.0.1, or 0.11.0.2) application instances </li>
     </ul>

     <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()</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/architecture" 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 API</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>
	<!--
	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 & API Changes</h1>

	<p>
	If you want to upgrade from 0.10.2.x to 0.11.0 you don't need to do any code changes as the public API is fully backward compatible.
	However, some configuration parameters were deprecated and thus it is recommend to update your code eventually to allow for future upgrades.
	See <a href="#streams_api_changes_0110">below</a> a complete list of 0.11.0 API and semantical changes that allow you to advance your application and/or simplify your code base, including the usage of new features.
	</p>

	<p>
	If you want to upgrade from 0.10.1.x to 0.11.0, also see the <a href="/{{version}}/upgrade/#upgrade_1020_streams"><b>Upgrade Section for 0.10.2</b></a>.
	It highlights incompatible changes you need to consider to upgrade your code and application.
	See below a complete list of <a href="#streams_api_changes_0102">0.10.2</a> and <a href="#streams_api_changes_0110">0.11.0</a> API and semantical changes
	that allow you to advance your application and/or simplify your code base, including the usage of new features.
	</p>

	<p>
	Upgrading from 0.10.0.x to 0.11.0.x directly is also possible.
	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.
	See <a href="#streams_api_changes_0101">Streams API changes in 0.10.1</a>, <a href="#streams_api_changes_0102">Streams API changes in 0.10.2</a>,
	and <a href="#streams_api_changes_0110">Streams API changes in 0.11.0</a> for a complete list of API changes.
	Upgrading to 0.11.0.3 requires two rolling bounces with config <code>upgrade.from="0.10.0"</code> set for first upgrade phase
	(cf. <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-268%3A+Simplify+Kafka+Streams+Rebalance+Metadata+Upgrade">KIP-268</a>).
	As an alternative, an offline upgrade is also possible.
	</p>
	<ul>
	<li> prepare your application instances for a rolling bounce and make sure that config <code>upgrade.from</code> is set to <code>"0.10.0"</code> for new version 0.11.0.3 </li>
	<li> bounce each instance of your application once </li>
	<li> prepare your newly deployed 0.11.0.3 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> Upgrading from 0.10.0.x to 0.11.0.0, 0.11.0.1, or 0.11.0.2 requires an offline upgrade (rolling bounce upgrade is not supported) </p>
	<ul>
	<li> stop all old (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 (0.11.0.0, 0.11.0.1, or 0.11.0.2) application instances </li>
	</ul>

	<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<? super K, ? super V> action)</code> </li>
	<li> <code>void print()</code> </li>
	<li> <code>void print(final String streamName)</code> </li>
	<li> <code>void print(final Serde<K> keySerde, final Serde<V> valSerde)</code> </li>
	<li> <code>void print(final Serde<K> keySerde, final Serde<V> 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<K> keySerde, final Serde<V> valSerde)</code> </li>
	<li> <code>void writeAsText(final String filePath, final String streamName, final Serde<K> keySerde, final Serde<V> 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<T> 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()</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/architecture" 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 API</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>