You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by lu...@apache.org on 2024/06/27 09:58:04 UTC
(kafka) branch 3.8 updated: MINOR: Update 3.8 documentation for Kafka Streams (#16265)
This is an automated email from the ASF dual-hosted git repository.
lucasbru pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.8 by this push:
new 5edab42ae37 MINOR: Update 3.8 documentation for Kafka Streams (#16265)
5edab42ae37 is described below
commit 5edab42ae371f40584b42fb9a9568cf0ab5c5037
Author: Lucas Brutschy <lb...@confluent.io>
AuthorDate: Wed Jun 26 13:55:18 2024 +0200
MINOR: Update 3.8 documentation for Kafka Streams (#16265)
All public interface changes should be briefly mentioned in the
upgrade guide.
Reviewers: Matthias J. Sax <ma...@confluent.io>, Anna Sophie Blee-Goldman <so...@responsive.dev>, Nick Telford <ni...@gmail.com>
---
docs/ops.html | 28 +++++++++++++++--
docs/streams/developer-guide/config-streams.html | 31 ++++++++++++++-----
docs/streams/upgrade-guide.html | 38 ++++++++++++++++++++++--
3 files changed, 84 insertions(+), 13 deletions(-)
diff --git a/docs/ops.html b/docs/ops.html
index 48a0e13e7a8..bb2270843b4 100644
--- a/docs/ops.html
+++ b/docs/ops.html
@@ -3201,9 +3201,11 @@ active-process-ratio metrics which have a recording level of <code>info</code>:
</table>
<h5 class="anchor-heading"><a id="kafka_streams_store_monitoring" class="anchor-link"></a><a href="#kafka_streams_store_monitoring">State Store Metrics</a></h5>
-All of the following metrics have a recording level of <code>debug</code>, except for the record-e2e-latency-* metrics which have a recording level <code>trace</code>.
-Note that the <code>store-scope</code> value is specified in <code>StoreSupplier#metricsScope()</code> for user's customized state stores;
-for built-in state stores, currently we have:
+All the following metrics have a recording level of <code>debug</code>, except for the <code>record-e2e-latency-*</code>
+metrics which have a recording level <code>trace</code> and <code>num-open-iterators</code> which has recording level
+<code>info</code>.
+Note that the <code>store-scope</code> value is specified in <code>StoreSupplier#metricsScope()</code> for user's
+customized state stores; for built-in state stores, currently we have:
<ul>
<li><code>in-memory-state</code></li>
<li><code>in-memory-lru-state</code></li>
@@ -3408,6 +3410,26 @@ for built-in state stores, currently we have:
<td>The minimum end-to-end latency of a record, measured by comparing the record timestamp with the system time when it has been fully processed by the node.</td>
<td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
</tr>
+ <tr>
+ <td>num-open-iterators</td>
+ <td>The current number of iterators on the store that have been created, but not yet closed.</td>
+ <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
+ </tr>
+ <tr>
+ <td>iterator-duration-avg</td>
+ <td>The average time spent between creating an iterator and closing it, in nanoseconds.</td>
+ <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
+ </tr>
+ <tr>
+ <td>iterator-duration-max</td>
+ <td>The maximum time spent between creating an iterator and closing it, in nanoseconds.</td>
+ <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
+ </tr>
+ <tr>
+ <td>oldest-iterator-open-since-ms</td>
+ <td>The UNIX timestamp the oldest still open iterator was created, in milliseconds.</td>
+ <td>kafka.streams:type=stream-state-metrics,thread-id=([-.\w]+),task-id=([-.\w]+),[store-scope]-id=([-.\w]+)</td>
+ </tr>
</tbody>
</table>
diff --git a/docs/streams/developer-guide/config-streams.html b/docs/streams/developer-guide/config-streams.html
index 86828518bdc..6d2c4049e72 100644
--- a/docs/streams/developer-guide/config-streams.html
+++ b/docs/streams/developer-guide/config-streams.html
@@ -90,6 +90,7 @@ settings.put(... , ...);</code></pre>
<li><a class="reference internal" href="#replication-factor" id="id13">replication.factor</a></li>
<li><a class="reference internal" href="#rocksdb-config-setter" id="id20">rocksdb.config.setter</a></li>
<li><a class="reference internal" href="#state-dir" id="id14">state.dir</a></li>
+ <li><a class="reference internal" href="#task-assignor-class" id="id39">task.assignor.class</a></li>
<li><a class="reference internal" href="#topology-optimization" id="id31">topology.optimization</a></li>
<li><a class="reference internal" href="#windowed-inner-class-serde" id="id38">windowed.inner.class.serde</a></li>
</ul>
@@ -333,7 +334,6 @@ streamsSettings.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);</code></pre>
the <code>org.apache.kafka.streams.state.DslStoreSuppliers</code> interface.
</td>
<td><code>BuiltInDslStoreSuppliers.RocksDBDslStoreSuppliers</code></td>
- <td>null</td>
</tr>
<tr class="row-odd"><td>max.task.idle.ms</td>
<td>Medium</td>
@@ -439,28 +439,33 @@ streamsSettings.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1);</code></pre>
<td colspan="2">Directory location for state stores.</td>
<td><code class="docutils literal"><span class="pre">/${java.io.tmpdir}/kafka-streams</span></code></td>
</tr>
- <tr class="row-even"><td>task.timeout.ms</td>
+ <tr class="row-even"><td>task.assignor.class</td>
+ <td>Medium</td>
+ <td colspan="2">A task assignor class or class name implementing the <code>TaskAssignor</code> interface.</td>
+ <td>The high-availability task assignor.</td>
+ </tr>
+ <tr class="row-odd"><td>task.timeout.ms</td>
<td>Medium</td>
<td colspan="2">The maximum amount of time in milliseconds a task might stall due to internal errors and retries until an error is raised. For a timeout of <code>0 ms</code>, a task would raise an error for the first internal error. For any timeout larger than <code>0 ms</code>, a task will retry at least once before an error is raised.</td>
<td>300000 milliseconds (5 minutes)</td>
</tr>
- <tr class="row-odd"><td>topology.optimization</td>
+ <tr class="row-even"><td>topology.optimization</td>
<td>Medium</td>
<td colspan="2">A configuration telling Kafka Streams if it should optimize the topology and what optimizations to apply. Acceptable values are: <code>StreamsConfig.NO_OPTIMIZATION</code> (<code>none</code>), <code>StreamsConfig.OPTIMIZE</code> (<code>all</code>) or a comma separated list of specific optimizations: <code>StreamsConfig.REUSE_KTABLE_SOURCE_TOPICS</code> (<code>reuse.ktable.source.topics</code>), <code>StreamsConfig.MERGE_REPARTITION_TOPICS</code> (<code>merge.r [...]
<code>StreamsConfig.SINGLE_STORE_SELF_JOIN</code> (<code>single.store.self.join</code>). </td>
- <td><code> NO_OPTIMIZATION</code></td>
+ <td><code>NO_OPTIMIZATION</code></td>
</tr>
- <tr class="row-even"><td>upgrade.from</td>
+ <tr class="row-odd"><td>upgrade.from</td>
<td>Medium</td>
<td colspan="2">The version you are upgrading from during a rolling upgrade.</td>
<td>See <a class="reference internal" href="#streams-developer-guide-upgrade-from"><span class="std std-ref">Upgrade From</span></a></td>
</tr>
- <tr class="row-odd"><td>windowstore.changelog.additional.retention.ms</td>
+ <tr class="row-even"><td>windowstore.changelog.additional.retention.ms</td>
<td>Low</td>
<td colspan="2">Added to a windows maintainMs to ensure data is not deleted from the log prematurely. Allows for clock drift.</td>
<td>86400000 milliseconds (1 day)</td>
</tr>
- <tr class="row-even"><td>window.size.ms</td>
+ <tr class="row-odd"><td>window.size.ms</td>
<td>Low</td>
<td colspan="2">Sets window size for the deserializer in order to calculate window end times.</td>
<td><code class="docutils literal"><span class="pre">null</span></code></td>
@@ -1017,6 +1022,18 @@ streamsConfig.put(StreamsConfig.ROCKSDB_CONFIG_SETTER_CLASS_CONFIG, CustomRocksD
this path must be unique for each such instance.</div>
</blockquote>
</div>
+ <div class="section" id="task-assignor-class">
+ <h4><a class="toc-backref" href="#id39">task.assignor.class</a><a class="headerlink" href="#task-assignor-class" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>A task assignor class or class name implementing the
+ <code>org.apache.kafka.streams.processor.assignment.TaskAssignor</code> interface. Defaults to the
+ high-availability task assignor. One possible alternative implementation provided in Apache Kafka is the
+ <code>org.apache.kafka.streams.processor.assignment.assignors.StickyTaskAssignor</code>, which was the default task
+ assignor before KIP-441 and minimizes task movement at the cost of stateful task availability. Alternative implementations of
+ the task assignment algorithm can be plugged into the application by implementing a custom <code>TaskAssignor</code> and setting this config to the name of the custom task assignor class.
+ </div>
+ </blockquote>
+ </div>
<div class="section" id="topology-optimization">
<h4><a class="toc-backref" href="#id31">topology.optimization</a><a class="headerlink" href="#topology-optimization" title="Permalink to this headline"></a></h4>
<blockquote>
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index 127cb99c3b1..f24edfbd42e 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -133,12 +133,44 @@
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>
- <h4><a id="streams_api_changes_380" href="#streams_api_changes_380">Streams API changes in 3.8.0</a></h3>
+ <h3><a id="streams_api_changes_380" href="#streams_api_changes_380">Streams API changes in 3.8.0</a></h3>
+
+ <p>
+ Kafka Streams now supports customizable task assignment strategies via the <code>task.assignor.class</code>
+ configuration. The configuration can be set to the fully qualified class name of a custom task assignor
+ implementation that has to extend the new
+ <code>org.apache.kafka.streams.processor.assignment.TaskAssignor</code> interface.
+
+ The new configuration also allows users to bring back the behavior of the old task assignor
+ <code>StickyTaskAssignor</code> that was used before the introduction of the
+ <code>HighAvailabilityTaskAssignor</code>. If no custom task assignor is configured, the default task assignor
+ <code>HighAvailabilityTaskAssignor</code> is used.
+
+ If you were using the <code>internal.task.assignor.class</code> config, you should switch to using the new
+ <code>task.assignor.class</code> config instead, as the internal config will be removed in a future release.
+ If you were previously plugging in the <code>StickyTaskAssignor</code> via the legacy
+ <code>internal.task.assignor.class</code> config, you will need to make sure that you are importing
+ the new <code>org.apache.kafka.streams.processor.assignment.StickTaskAssignor</code> when you switch
+ over to the new <code>task.assignor.class</code> config, which is a version of the <code>StickyTaskAssignor</code>
+ that implements the new public <code>TaskAssignor</code> interface.
+
+ For more details, see the public interface section of
+ <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-924%3A+customizable+task+assignment+for+Streams">KIP-924</a>.
+ </p>
+
<p>
The Processor API now support so-called read-only state stores, added via
<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-813%3A+Shareable+State+Stores">KIP-813</a>.
- These stores don't have a dedicated changelog topic, but use their source topic for fault-tolerance,
- simlar to <code>KTables</code> with source-topic optimization enabled.
+ These stores don't have a dedicated changelog topic, but use their source topic for fault-tolerance,
+ similar to <code>KTables</code> with source-topic optimization enabled.
+ </p>
+
+ <p>
+ To improve detection of leaked state store iterators, we added new store-level metrics to track the number and
+ age of open iterators. The new metrics are <code>num-open-iterators</code>, <code>iterator-duration-avg</code>,
+ <code>iterator-duration-max</code> and <code>oldest-iterator-open-since-ms</code>. These metrics are available
+ for all state stores, including RocksDB, in-memory, and custom stores. More details can be found in
+ <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-989%3A+Improved+StateStore+Iterator+metrics+for+detecting+leaks">KIP-989</a>.
</p>
<h3><a id="streams_api_changes_370" href="#streams_api_changes_370">Streams API changes in 3.7.0</a></h3>