You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by vv...@apache.org on 2020/07/21 02:31:52 UTC
[kafka-site] branch asf-site updated: MINOR: update Streams docs
for 2.6 and fix configs (#275)
This is an automated email from the ASF dual-hosted git repository.
vvcephei pushed a commit to branch asf-site
in repository https://gitbox.apache.org/repos/asf/kafka-site.git
The following commit(s) were added to refs/heads/asf-site by this push:
new 4c19eab MINOR: update Streams docs for 2.6 and fix configs (#275)
4c19eab is described below
commit 4c19eab4f6315c91549be0bff08ad46173be4f64
Author: A. Sophie Blee-Goldman <so...@confluent.io>
AuthorDate: Mon Jul 20 19:31:40 2020 -0700
MINOR: update Streams docs for 2.6 and fix configs (#275)
Add docs for KIP-441 and KIP-613.
Fixed some miscellaneous unrelated issues in the docs:
* Adds some missing configs to the Streams config docs: max.task.idle.ms,topology.optimization, default.windowed.key.serde.inner.class, and default.windowed.value.serde.inner.class
* Defines the previously-undefined default windowed serde class configs, including choosing a default (null) and giving them a doc string, so the yshould nwo show up in the auto-generated general Kafka config docs
* Adds a note to warn users about the rocksDB bug that prevents setting a strict capacity limit and counting write buffer memory against the block cache
Ports: https://github.com/apache/kafka/pull/9027
Reviewers: John Roesler <vv...@apache.org>
---
26/streams/architecture.html | 4 +-
26/streams/developer-guide/config-streams.html | 772 +++++++++++++++----------
26/streams/developer-guide/memory-mgmt.html | 5 +-
26/streams/developer-guide/running-app.html | 12 +
26/streams/upgrade-guide.html | 16 +-
5 files changed, 485 insertions(+), 324 deletions(-)
diff --git a/26/streams/architecture.html b/26/streams/architecture.html
index 3ef9dfe..43de9e7 100644
--- a/26/streams/architecture.html
+++ b/26/streams/architecture.html
@@ -151,8 +151,10 @@
<p>
Note that the cost of task (re)initialization typically depends primarily on the time for restoring the state by replaying the state stores' associated changelog topics.
To minimize this restoration time, users can configure their applications to have <b>standby replicas</b> of local states (i.e. fully replicated copies of the state).
- When a task migration happens, Kafka Streams then attempts to assign a task to an application instance where such a standby replica already exists in order to minimize
+ When a task migration happens, Kafka Streams will assign a task to an application instance where such a standby replica already exists in order to minimize
the task (re)initialization cost. See <code>num.standby.replicas</code> in the <a href="/{{version}}/documentation/#streamsconfigs"><b>Kafka Streams Configs</b></a> section.
+ Starting in 2.6, Kafka Streams will guarantee that a task is only ever assigned to an instance with a fully caught-up local copy of the state, if such an instance
+ exists. Standby tasks will increase the likelihood that a caught-up instance exists in the case of a failure.
</p>
<div class="pagination">
diff --git a/26/streams/developer-guide/config-streams.html b/26/streams/developer-guide/config-streams.html
index a64dd77..77fe8a4 100644
--- a/26/streams/developer-guide/config-streams.html
+++ b/26/streams/developer-guide/config-streams.html
@@ -63,24 +63,31 @@
</ul>
</li>
<li><a class="reference internal" href="#optional-configuration-parameters" id="id6">Optional configuration parameters</a><ul>
+ <li><a class="reference internal" href="#acceptable-recovery-lag" id="id27">acceptable.recovery.lag</a></li>
<li><a class="reference internal" href="#default-deserialization-exception-handler" id="id7">default.deserialization.exception.handler</a></li>
- <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
<li><a class="reference internal" href="#default-key-serde" id="id8">default.key.serde</a></li>
+ <li><a class="reference internal" href="#default-production-exception-handler" id="id24">default.production.exception.handler</a></li>
+ <li><a class="reference internal" href="#timestamp-extractor" id="id15">default.timestamp.extractor</a></li>
<li><a class="reference internal" href="#default-value-serde" id="id9">default.value.serde</a></li>
+ <li><a class="reference internal" href="#default-windowed-key-serde-inner" id="id32">default.windowed.key.serde.inner</a></li>
+ <li><a class="reference internal" href="#default-windowed-value-serde-inner" id="id33">default.windowed.value.serde.inner</a></li>
+ <li><a class="reference internal" href="#max-task-idle-ms" id="id28">max.task.idle.ms</a></li>
+ <li><a class="reference internal" href="#max-warmup-replicas" id="id29">max.warmup.replicas</a></li>
<li><a class="reference internal" href="#num-standby-replicas" id="id10">num.standby.replicas</a></li>
<li><a class="reference internal" href="#num-stream-threads" id="id11">num.stream.threads</a></li>
<li><a class="reference internal" href="#partition-grouper" id="id12">partition.grouper</a></li>
+ <li><a class="reference internal" href="#probing-rebalance-interval-ms" id="id30">probing.rebalance.interval.ms</a></li>
<li><a class="reference internal" href="#processing-guarantee" id="id25">processing.guarantee</a></li>
<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="#timestamp-extractor" id="id15">timestamp.extractor</a></li>
+ <li><a class="reference internal" href="#topology-optimization" id="id31">topology.optimization</a></li>
</ul>
</li>
<li><a class="reference internal" href="#kafka-consumers-and-producer-configuration-parameters" id="id16">Kafka consumers and producer configuration parameters</a><ul>
<li><a class="reference internal" href="#naming" id="id17">Naming</a></li>
<li><a class="reference internal" href="#default-values" id="id18">Default Values</a></li>
<li><a class="reference internal" href="#enable-auto-commit" id="id19">enable.auto.commit</a></li>
- <li><a class="reference internal" href="#rocksdb-config-setter" id="id20">rocksdb.config.setter</a></li>
</ul>
</li>
<li><a class="reference internal" href="#recommended-configuration-parameters-for-resiliency" id="id21">Recommended configuration parameters for resiliency</a><ul>
@@ -166,6 +173,11 @@
</tr>
</thead>
<tbody valign="top">
+ <tr class="row-odd"><td>acceptable.recovery.lag</td>
+ <td>Medium</td>
+ <td colspan="2">The maximum acceptable lag (number of offsets to catch up) for an instance to be considered caught-up and ready for the active task.</td>
+ <td>10000</td>
+ </tr>
<tr class="row-even"><td>application.server</td>
<td>Low</td>
<td colspan="2">A host:port pair pointing to an embedded user defined endpoint that can be used for discovering the locations of
@@ -199,16 +211,47 @@
<td colspan="2">Exception handling class that implements the <code class="docutils literal"><span class="pre">DeserializationExceptionHandler</span></code> interface.</td>
<td><code class="docutils literal"><span class="pre">LogAndContinueExceptionHandler</span></code></td>
</tr>
- <tr class="row-even"><td>default.production.exception.handler</td>
+ <tr class="row-even"><td>default.key.serde</td>
+ <td>Medium</td>
+ <td colspan="2">Default serializer/deserializer class for record keys, implements the <code class="docutils literal"><span class="pre">Serde</span></code> interface (see also default.value.serde).</td>
+ <td><code class="docutils literal"><span class="pre">Serdes.ByteArray().getClass().getName()</span></code></td>
+ </tr>
+ <tr class="row-odd"><td>default.production.exception.handler</td>
<td>Medium</td>
<td colspan="2">Exception handling class that implements the <code class="docutils literal"><span class="pre">ProductionExceptionHandler</span></code> interface.</td>
<td><code class="docutils literal"><span class="pre">DefaultProductionExceptionHandler</span></code></td>
</tr>
- <tr class="row-odd"><td>key.serde</td>
+ <tr class="row-even"><td>default.timestamp.extractor</td>
+ <td>Medium</td>
+ <td colspan="2">Timestamp extractor class that implements the <code class="docutils literal"><span class="pre">TimestampExtractor</span></code> interface.</td>
+ <td>See <a class="reference internal" href="#streams-developer-guide-timestamp-extractor"><span class="std std-ref">Timestamp Extractor</span></a></td>
+ </tr>
+ <tr class="row-odd"><td>default.value.serde</td>
<td>Medium</td>
- <td colspan="2">Default serializer/deserializer class for record keys, implements the <code class="docutils literal"><span class="pre">Serde</span></code> interface (see also value.serde).</td>
+ <td colspan="2">Default serializer/deserializer class for record values, implements the <code class="docutils literal"><span class="pre">Serde</span></code> interface (see also default.key.serde).</td>
<td><code class="docutils literal"><span class="pre">Serdes.ByteArray().getClass().getName()</span></code></td>
</tr>
+ <tr class="row-even"><td>default.windowed.key.serde.inner</td>
+ <td>Medium</td>
+ <td colspan="2">Default serializer/deserializer for the inner class of windowed keys, implementing the <code class="docutils literal"><span class="pre">Serde</span></code> interface.</td>
+ <td>null</td>
+ </tr>
+ <tr class="row-odd"><td>default.windowed.value.serde.inner</td>
+ <td>Medium</td>
+ <td colspan="2">Default serializer/deserializer for the inner class of windowed values, implementing the <code class="docutils literal"><span class="pre">Serde</span></code> interface.</td>
+ <td>null</td>
+ </tr>
+ <tr class="row-even"><td>max.task.idle.ms</td>
+ <td>Medium</td>
+ <td colspan="2">Maximum amount of time a stream task will stay idle while waiting for all partitions to contain data and avoid potential out-of-order record
+ processing across multiple input streams.</td>
+ <td>0 milliseconds</td>
+ </tr>
+ <tr class="row-odd"><td>max.warmup.replicas</td>
+ <td>Medium</td>
+ <td colspan="2">The maximum number of warmup replicas (extra standbys beyond the configured num.standbys) that can be assigned at once.</td>
+ <td>2</td>
+ </tr>
<tr class="row-even"><td>metric.reporters</td>
<td>Low</td>
<td colspan="2">A list of classes to use as metrics reporters.</td>
@@ -244,10 +287,15 @@
<td colspan="2">Partition grouper class that implements the <code class="docutils literal"><span class="pre">PartitionGrouper</span></code> interface.</td>
<td>See <a class="reference internal" href="#streams-developer-guide-partition-grouper"><span class="std std-ref">Partition Grouper</span></a></td>
</tr>
+ <tr class="row-odd"><td>probing.rebalance.interval.ms</td>
+ <td>Low</td>
+ <td colspan="2">The maximum time to wait before triggering a rebalance to probe for warmup replicas that have sufficiently caught up.</td>
+ <td>600000 milliseconds (10 minutes)</td>
+ </tr>
<tr class="row-even"><td>processing.guarantee</td>
<td>Medium</td>
<td colspan="2">The processing mode. Can be either <code class="docutils literal"><span class="pre">"at_least_once"</span></code> (default),
- <code class="docutils literal"><span class="pre">"exactly_once"</span></code>, or <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code>.
+ <code class="docutils literal"><span class="pre">"exactly_once"</span></code>, or <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code></td>.
<td>See <a class="reference internal" href="#streams-developer-guide-processing-guarantedd"><span class="std std-ref">Processing Guarantee</span></a></td>
</tr>
<tr class="row-odd"><td>poll.ms</td>
@@ -260,46 +308,41 @@
<td colspan="2">The replication factor for changelog topics and repartition topics created by the application.</td>
<td>1</td>
</tr>
- <tr class="row-even"><td>retries</td>
- <td>Medium</td>
- <td colspan="2">The number of retries for broker requests that return a retryable error. </td>
- <td>0</td>
+ <tr class="row-odd"><td>retries</td>
+ <td>Medium</td>
+ <td colspan="2">The number of retries for broker requests that return a retryable error. </td>
+ <td>0</td>
</tr>
<tr class="row-even"><td>retry.backoff.ms</td>
- <td>Medium</td>
- <td colspan="2">The amount of time in milliseconds, before a request is retried. This applies if the <code class="docutils literal"><span class="pre">retries</span></code> parameter is configured to be greater than 0. </td>
- <td>100</td>
+ <td>Medium</td>
+ <td colspan="2">The amount of time in milliseconds, before a request is retried. This applies if the <code class="docutils literal"><span class="pre">retries</span></code> parameter is configured to be greater than 0. </td>
+ <td>100</td>
</tr>
- <tr class="row-even"><td>rocksdb.config.setter</td>
+ <tr class="row-odd"><td>rocksdb.config.setter</td>
<td>Medium</td>
<td colspan="2">The RocksDB configuration.</td>
<td></td>
</tr>
- <tr class="row-odd"><td>state.cleanup.delay.ms</td>
+ <tr class="row-even"><td>state.cleanup.delay.ms</td>
<td>Low</td>
<td colspan="2">The amount of time in milliseconds to wait before deleting state when a partition has migrated.</td>
<td>600000 milliseconds</td>
</tr>
- <tr class="row-even"><td>state.dir</td>
+ <tr class="row-odd"><td>state.dir</td>
<td>High</td>
<td colspan="2">Directory location for state stores.</td>
<td><code class="docutils literal"><span class="pre">/tmp/kafka-streams</span></code></td>
</tr>
- <tr class="row-odd"><td>timestamp.extractor</td>
+ <tr class="row-even"><td>topology.optimization</td>
<td>Medium</td>
- <td colspan="2">Timestamp extractor class that implements the <code class="docutils literal"><span class="pre">TimestampExtractor</span></code> interface.</td>
- <td>See <a class="reference internal" href="#streams-developer-guide-timestamp-extractor"><span class="std std-ref">Timestamp Extractor</span></a></td>
+ <td colspan="2">A configuration telling Kafka Streams if it should optimize the topology</td>
+ <td>none</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>value.serde</td>
- <td>Medium</td>
- <td colspan="2">Default serializer/deserializer class for record values, implements the <code class="docutils literal"><span class="pre">Serde</span></code> interface (see also key.serde).</td>
- <td><code class="docutils literal"><span class="pre">Serdes.ByteArray().getClass().getName()</span></code></td>
- </tr>
<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>
@@ -307,6 +350,22 @@
</tr>
</tbody>
</table>
+ <div class="section" id="acceptable-recovery-lag">
+ <h4><a class="toc-backref" href="#id27">acceptable.recovery.lag</a><a class="headerlink" href="#acceptable-recovery-lag" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>
+ <p>
+ The maximum acceptable lag (total number of offsets to catch up from the changelog) for an instance to be considered caught-up and able to receive an active task. Streams will only assign
+ stateful active tasks to instances whose state stores are within the acceptable recovery lag, if any exist, and assign warmup replicas to restore state in the background for instances
+ that are not yet caught up. Should correspond to a recovery time of well under a minute for a given workload. Must be at least 0.
+ </p>
+ <p>
+ Note: if you set this to <code>Long.MAX_VALUE</code> it effectively disables the warmup replicas and task high availability, allowing Streams to immediately produce a balanced
+ assignment and migrate tasks to a new instance without first warming them up.
+ </p>
+ </div>
+ </blockquote>
+ </div>
<div class="section" id="default-deserialization-exception-handler">
<span id="streams-developer-guide-deh"></span><h4><a class="toc-backref" href="#id7">default.deserialization.exception.handler</a><a class="headerlink" href="#default-deserialization-exception-handler" title="Permalink to this headline"></a></h4>
<blockquote>
@@ -368,10 +427,10 @@
such as attempting to produce a record that is too large. By default, Kafka provides and uses the <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/errors/DefaultProductionExceptionHandler.html">DefaultProductionExceptionHandler</a>
that always fails when these exceptions occur.</p>
- <p>Each exception handler can return a <code>FAIL</code> or <code>CONTINUE</code> depending on the record and the exception thrown. Returning <code>FAIL</code> will signal that Streams should shut down and <code>CONTINUE</code> will signal that Streams
- should ignore the issue and continue processing. If you want to provide an exception handler that always ignores records that are too large, you could implement something like the following:</p>
+ <p>Each exception handler can return a <code>FAIL</code> or <code>CONTINUE</code> depending on the record and the exception thrown. Returning <code>FAIL</code> will signal that Streams should shut down and <code>CONTINUE</code> will signal that Streams
+ should ignore the issue and continue processing. If you want to provide an exception handler that always ignores records that are too large, you could implement something like the following:</p>
- <pre class="brush: java;">
+ <pre class="brush: java;">
import java.util.Properties;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.common.errors.RecordTooLargeException;
@@ -399,18 +458,106 @@
IgnoreRecordTooLargeHandler.class);</pre></div>
</blockquote>
</div>
+ <div class="section" id="timestamp-extractor">
+ <span id="streams-developer-guide-timestamp-extractor"></span><h4><a class="toc-backref" href="#id15">default.timestamp.extractor</a><a class="headerlink" href="#timestamp-extractor" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div><p>A timestamp extractor pulls a timestamp from an instance of <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html">ConsumerRecord</a>.
+ Timestamps are used to control the progress of streams.</p>
+ <p>The default extractor is
+ <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/FailOnInvalidTimestamp.html">FailOnInvalidTimestamp</a>.
+ This extractor retrieves built-in timestamps that are automatically embedded into Kafka messages by the Kafka producer
+ client since
+ <a class="reference external" href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message">Kafka version 0.10</a>.
+ Depending on the setting of Kafka’s server-side <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> broker and <code class="docutils literal"><span class="pre">message.timestamp.type</span></code> topic parameters,
+ this extractor provides you with:</p>
+ <ul class="simple">
+ <li><strong>event-time</strong> processing semantics if <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> is set to <code class="docutils literal"><span class="pre">CreateTime</span></code> aka “producer time”
+ (which is the default). This represents the time when a Kafka producer sent the original message. If you use Kafka’s
+ official producer client, the timestamp represents milliseconds since the epoch.</li>
+ <li><strong>ingestion-time</strong> processing semantics if <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> is set to <code class="docutils literal"><span class="pre">LogAppendTime</span></code> aka “broker
+ time”. This represents the time when the Kafka broker received the original message, in milliseconds since the epoch.</li>
+ </ul>
+ <p>The <code class="docutils literal"><span class="pre">FailOnInvalidTimestamp</span></code> extractor throws an exception if a record contains an invalid (i.e. negative) built-in
+ timestamp, because Kafka Streams would not process this record but silently drop it. Invalid built-in timestamps can
+ occur for various reasons: if for example, you consume a topic that is written to by pre-0.10 Kafka producer clients
+ or by third-party producer clients that don’t support the new Kafka 0.10 message format yet; another situation where
+ this may happen is after upgrading your Kafka cluster from <code class="docutils literal"><span class="pre">0.9</span></code> to <code class="docutils literal"><span class="pre">0.10</span></code>, where all the data that was generated
+ with <code class="docutils literal"><span class="pre">0.9</span></code> does not include the <code class="docutils literal"><span class="pre">0.10</span></code> message timestamps.</p>
+ <p>If you have data with invalid timestamps and want to process it, then there are two alternative extractors available.
+ Both work on built-in timestamps, but handle invalid timestamps differently.</p>
+ <ul class="simple">
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/LogAndSkipOnInvalidTimestamp.html">LogAndSkipOnInvalidTimestamp</a>:
+ This extractor logs a warn message and returns the invalid timestamp to Kafka Streams, which will not process but
+ silently drop the record.
+ This log-and-skip strategy allows Kafka Streams to make progress instead of failing if there are records with an
+ invalid built-in timestamp in your input data.</li>
+ <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/UsePartitionTimeOnInvalidTimestamp.html">UsePartitionTimeOnInvalidTimestamp</a>.
+ This extractor returns the record’s built-in timestamp if it is valid (i.e. not negative). If the record does not
+ have a valid built-in timestamps, the extractor returns the previously extracted valid timestamp from a record of the
+ same topic partition as the current record as a timestamp estimation. In case that no timestamp can be estimated, it
+ throws an exception.</li>
+ </ul>
+ <p>Another built-in extractor is
+ <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/WallclockTimestampExtractor.html">WallclockTimestampExtractor</a>.
+ This extractor does not actually “extract” a timestamp from the consumed record but rather returns the current time in
+ milliseconds from the system clock (think: <code class="docutils literal"><span class="pre">System.currentTimeMillis()</span></code>), which effectively means Streams will operate
+ on the basis of the so-called <strong>processing-time</strong> of events.</p>
+ <p>You can also provide your own timestamp extractors, for instance to retrieve timestamps embedded in the payload of
+ messages. If you cannot extract a valid timestamp, you can either throw an exception, return a negative timestamp, or
+ estimate a timestamp. Returning a negative timestamp will result in data loss – the corresponding record will not be
+ processed but silently dropped. If you want to estimate a new timestamp, you can use the value provided via
+ <code class="docutils literal"><span class="pre">previousTimestamp</span></code> (i.e., a Kafka Streams timestamp estimation). Here is an example of a custom
+ <code class="docutils literal"><span class="pre">TimestampExtractor</span></code> implementation:</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="kn">import</span> <span class="nn">org.apache.kafka.clients.consumer.ConsumerRecord</span><span class="o">;</span>
+<span class="kn">import</span> <span class="nn">org.apache.kafka.streams.processor.TimestampExtractor</span><span class="o">;</span>
+
+<span class="c1">// Extracts the embedded timestamp of a record (giving you "event-time" semantics).</span>
+<span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyEventTimeExtractor</span> <span class="kd">implements</span> <span class="n">TimestampExtractor</span> <span class="o">{</span>
+
+ <span class="nd">@Override</span>
+ <span class="kd">public</span> <span class="kt">long</span> <span class="nf">extract</span><span class="o">(</span><span class="kd">final</span> <span class="n">ConsumerRecord</span><span class="o"><</span><span class="n">Object</span><span class="o">,</span> <span class="n">Object</span><span class="o">></span> <span class="n">record</span><span class="o">,</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">previousTimestamp</span><span class="o">) [...]
+ <span class="c1">// `Foo` is your own custom class, which we assume has a method that returns</span>
+ <span class="c1">// the embedded timestamp (milliseconds since midnight, January 1, 1970 UTC).</span>
+ <span class="kt">long</span> <span class="n">timestamp</span> <span class="o">=</span> <span class="o">-</span><span class="mi">1</span><span class="o">;</span>
+ <span class="kd">final</span> <span class="n">Foo</span> <span class="n">myPojo</span> <span class="o">=</span> <span class="o">(</span><span class="n">Foo</span><span class="o">)</span> <span class="n">record</span><span class="o">.</span><span class="na">value</span><span class="o">();</span>
+ <span class="k">if</span> <span class="o">(</span><span class="n">myPojo</span> <span class="o">!=</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
+ <span class="n">timestamp</span> <span class="o">=</span> <span class="n">myPojo</span><span class="o">.</span><span class="na">getTimestampInMillis</span><span class="o">();</span>
+ <span class="o">}</span>
+ <span class="k">if</span> <span class="o">(</span><span class="n">timestamp</span> <span class="o"><</span> <span class="mi">0</span><span class="o">)</span> <span class="o">{</span>
+ <span class="c1">// Invalid timestamp! Attempt to estimate a new timestamp,</span>
+ <span class="c1">// otherwise fall back to wall-clock time (processing-time).</span>
+ <span class="k">if</span> <span class="o">(</span><span class="n">previousTimestamp</span> <span class="o">>=</span> <span class="mi">0</span><span class="o">)</span> <span class="o">{</span>
+ <span class="k">return</span> <span class="n">previousTimestamp</span><span class="o">;</span>
+ <span class="o">}</span> <span class="k">else</span> <span class="o">{</span>
+ <span class="k">return</span> <span class="n">System</span><span class="o">.</span><span class="na">currentTimeMillis</span><span class="o">();</span>
+ <span class="o">}</span>
+ <span class="o">}</span>
+ <span class="o">}</span>
+
+<span class="o">}</span>
+</pre></div>
+ </div>
+ <p>You would then define the custom timestamp extractor in your Streams configuration as follows:</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="kn">import</span> <span class="nn">java.util.Properties</span><span class="o">;</span>
+<span class="kn">import</span> <span class="nn">org.apache.kafka.streams.StreamsConfig</span><span class="o">;</span>
+
+<span class="n">Properties</span> <span class="n">streamsConfiguration</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+<span class="n">streamsConfiguration</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG</span><span class="o">,</span> <span class="n">MyEventTimeExtractor</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
+</pre></div>
+ </div>
+ </div></blockquote>
+ </div>
<div class="section" id="default-key-serde">
<h4><a class="toc-backref" href="#id8">default.key.serde</a><a class="headerlink" href="#default-key-serde" title="Permalink to this headline"></a></h4>
<blockquote>
<div><p>The default Serializer/Deserializer class for record keys. Serialization and deserialization in Kafka Streams happens
whenever data needs to be materialized, for example:</p>
- <blockquote>
<div><ul class="simple">
<li>Whenever data is read from or written to a <em>Kafka topic</em> (e.g., via the <code class="docutils literal"><span class="pre">StreamsBuilder#stream()</span></code> and <code class="docutils literal"><span class="pre">KStream#to()</span></code> methods).</li>
<li>Whenever data is read from or written to a <em>state store</em>.</li>
</ul>
<p>This is discussed in more detail in <a class="reference internal" href="datatypes.html#streams-developer-guide-serdes"><span class="std std-ref">Data types and serialization</span></a>.</p>
- </div></blockquote>
+ </div>
</div></blockquote>
</div>
<div class="section" id="default-value-serde">
@@ -425,6 +572,51 @@
<p>This is discussed in more detail in <a class="reference internal" href="datatypes.html#streams-developer-guide-serdes"><span class="std std-ref">Data types and serialization</span></a>.</p>
</div></blockquote>
</div>
+ <div class="section" id="default-windowed-key-serde-inner">
+ <h4><a class="toc-backref" href="#id32">default.windowed.key.serde.inner</a><a class="headerlink" href="#default-windowed-key-serde-inner" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div><p>The default Serializer/Deserializer class for the inner class of windowed keys. Serialization and deserialization in Kafka Streams happens
+ whenever data needs to be materialized, for example:</p>
+ <div><ul class="simple">
+ <li>Whenever data is read from or written to a <em>Kafka topic</em> (e.g., via the <code class="docutils literal"><span class="pre">StreamsBuilder#stream()</span></code> and <code class="docutils literal"><span class="pre">KStream#to()</span></code> methods).</li>
+ <li>Whenever data is read from or written to a <em>state store</em>.</li>
+ </ul>
+ <p>This is discussed in more detail in <a class="reference internal" href="datatypes.html#streams-developer-guide-serdes"><span class="std std-ref">Data types and serialization</span></a>.</p>
+ </div>
+ </div></blockquote>
+ </div>
+ <div class="section" id="default-windowed-value-serde-inner">
+ <h4><a class="toc-backref" href="#id33">default.windowed.value.serde.inner</a><a class="headerlink" href="#default-windowed-value-serde-inner" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div><p>The default Serializer/Deserializer class for the inner class of windowed values. Serialization and deserialization in Kafka Streams happens
+ happens whenever data needs to be materialized, for example:</p>
+ <ul class="simple">
+ <li>Whenever data is read from or written to a <em>Kafka topic</em> (e.g., via the <code class="docutils literal"><span class="pre">StreamsBuilder#stream()</span></code> and <code class="docutils literal"><span class="pre">KStream#to()</span></code> methods).</li>
+ <li>Whenever data is read from or written to a <em>state store</em>.</li>
+ </ul>
+ <p>This is discussed in more detail in <a class="reference internal" href="datatypes.html#streams-developer-guide-serdes"><span class="std std-ref">Data types and serialization</span></a>.</p>
+ </div></blockquote>
+ </div>
+ <div class="section" id="max-task-idle-ms">
+ <span id="streams-developer-guide-max-task-idle-ms"></span><h4><a class="toc-backref" href="#id28">max.task.idle.ms</a><a class="headerlink" href="#max-task-idle-ms" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>
+ The maximum amount of time a task will idle without processing data when waiting for all of its input partition buffers to contain records. This can help avoid potential out-of-order
+ processing when the task has multiple input streams, as in a join, for example. Setting this to a nonzero value may increase latency but will improve time synchronization.
+ </div>
+ </blockquote>
+ </div>
+ <div class="section" id="max-warmup-replicas">
+ <span id="streams-developer-guide-max-warmup-replicas"></span><h4><a class="toc-backref" href="#id29">max.warmup.replicas</a><a class="headerlink" href="#max-warmup-replicas" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>
+ The maximum number of warmup replicas (extra standbys beyond the configured num.standbys) that can be assigned at once for the purpose of keeping
+ the task available on one instance while it is warming up on another instance it has been reassigned to. Used to throttle how much extra broker
+ traffic and cluster state can be used for high availability. Increasing this will allow Streams to warm up more tasks at once, speeding up the time
+ for the reassigned warmups to restore sufficient state for them to be transitioned to active tasks. Must be at least 1.
+ </div>
+ </blockquote>
+ </div>
<div class="section" id="num-standby-replicas">
<span id="streams-developer-guide-standby-replicas"></span><h4><a class="toc-backref" href="#id10">num.standby.replicas</a><a class="headerlink" href="#num-standby-replicas" title="Permalink to this headline"></a></h4>
<blockquote>
@@ -433,13 +625,15 @@
Standby replicas are used to minimize the latency of task failover. A task that was previously running on a failed instance is
preferred to restart on an instance that has standby replicas so that the local state store restoration process from its
changelog can be minimized. Details about how Kafka Streams makes use of the standby replicas to minimize the cost of
- resuming tasks on failover can be found in the <a class="reference internal" href="../architecture.html#streams_architecture_state"><span class="std std-ref">State</span></a> section.</div></blockquote>
+ resuming tasks on failover can be found in the <a class="reference internal" href="../architecture.html#streams_architecture_state"><span class="std std-ref">State</span></a> section.
</div>
- <div class="admonition note">
- <p class="first admonition-title">Note</p>
- <p class="last">If you enable <cite>n</cite> standby tasks, you need to provision <cite>n+1</cite> <code class="docutils literal"><span class="pre">KafkaStreams</span></code>
- instances.</p>
- </div>
+ </blockquote>
+ </div>
+ <div class="admonition note">
+ <p class="first admonition-title">Note</p>
+ <p class="last">If you enable <cite>n</cite> standby tasks, you need to provision <cite>n+1</cite> <code class="docutils literal"><span class="pre">KafkaStreams</span></code>
+ instances.</p>
+ </div>
<div class="section" id="num-stream-threads">
<h4><a class="toc-backref" href="#id11">num.stream.threads</a><a class="headerlink" href="#num-stream-threads" title="Permalink to this headline"></a></h4>
<blockquote>
@@ -449,31 +643,43 @@
<div class="section" id="partition-grouper">
<span id="streams-developer-guide-partition-grouper"></span><h4><a class="toc-backref" href="#id12">partition.grouper</a><a class="headerlink" href="#partition-grouper" title="Permalink to this headline"></a></h4>
<blockquote>
- <div>A partition grouper creates a list of stream tasks from the partitions of source topics, where each created task is assigned with a group of source topic partitions.
+ <div>
+ <b>[DEPRECATED]</b> A partition grouper creates a list of stream tasks from the partitions of source topics, where each created task is assigned with a group of source topic partitions.
The default implementation provided by Kafka Streams is <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/DefaultPartitionGrouper.html">DefaultPartitionGrouper</a>.
It assigns each task with one partition for each of the source topic partitions. The generated number of tasks equals the largest
- number of partitions among the input topics. Usually an application does not need to customize the partition grouper.</div></blockquote>
+ number of partitions among the input topics. Usually an application does not need to customize the partition grouper.
+ </div>
+ </blockquote>
+ </div>
+ <div class="section" id="probing-rebalance-interval-ms">
+ <h4><a class="toc-backref" href="#id30">probing.rebalance.interval.ms</a><a class="headerlink" href="#probing-rebalance-interval-ms" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>
+ The maximum time to wait before triggering a rebalance to probe for warmup replicas that have restored enough to be considered caught up. Streams will only assign stateful active tasks to
+ instances that are caught up and within the <a class="reference internal" href="#acceptable-recovery-lag"><span class="std std-ref">acceptable.recovery.lag</span></a>, if any exist. Probing rebalances are used to query the latest total lag of warmup replicas and transition
+ them to active tasks if ready. They will continue to be triggered as long as there are warmup tasks, and until the assignment is balanced. Must be at least 1 minute.
+ </div></blockquote>
</div>
<div class="section" id="processing-guarantee">
<span id="streams-developer-guide-processing-guarantee"></span><h4><a class="toc-backref" href="#id25">processing.guarantee</a><a class="headerlink" href="#processing-guarantee" title="Permalink to this headline"></a></h4>
<blockquote>
<div>The processing guarantee that should be used.
- Possible values are <code class="docutils literal"><span class="pre">"at_least_once"</span></code> (default),
- <code class="docutils literal"><span class="pre">"exactly_once"</span></code>,
- and <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code>.
- Using <code class="docutils literal"><span class="pre">"exactly_once"</span></code> requires broker
- version 0.11.0 or newer, while using <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code>
- requires broker version 2.5 or newer.
- Note that if exactly-once processing is enabled, the default for parameter
- <code class="docutils literal"><span class="pre">commit.interval.ms</span></code> changes to 100ms.
- Additionally, consumers are configured with <code class="docutils literal"><span class="pre">isolation.level="read_committed"</span></code>
- and producers are configured with <code class="docutils literal"><span class="pre">enable.idempotence=true</span></code> per default.
- Note that by default exactly-once processing requires a cluster of at least three brokers what is the recommended setting for production.
- For development, you can change this configuration by adjusting broker setting
- <code class="docutils literal"><span class="pre">transaction.state.log.replication.factor</span></code>
- and <code class="docutils literal"><span class="pre">transaction.state.log.min.isr</span></code>
- to the number of brokers you want to use.
- For more details see <a href="../core-concepts#streams_processing_guarantee">Processing Guarantees</a>.
+ Possible values are <code class="docutils literal"><span class="pre">"at_least_once"</span></code> (default),
+ <code class="docutils literal"><span class="pre">"exactly_once"</span></code>,
+ and <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code>.
+ Using <code class="docutils literal"><span class="pre">"exactly_once"</span></code> requires broker
+ version 0.11.0 or newer, while using <code class="docutils literal"><span class="pre">"exactly_once_beta"</span></code>
+ requires broker version 2.5 or newer.
+ Note that if exactly-once processing is enabled, the default for parameter
+ <code class="docutils literal"><span class="pre">commit.interval.ms</span></code> changes to 100ms.
+ Additionally, consumers are configured with <code class="docutils literal"><span class="pre">isolation.level="read_committed"</span></code>
+ and producers are configured with <code class="docutils literal"><span class="pre">enable.idempotence=true</span></code> per default.
+ Note that by default exactly-once processing requires a cluster of at least three brokers what is the recommended setting for production.
+ For development, you can change this configuration by adjusting broker setting
+ <code class="docutils literal"><span class="pre">transaction.state.log.replication.factor</span></code>
+ and <code class="docutils literal"><span class="pre">transaction.state.log.min.isr</span></code>
+ to the number of brokers you want to use.
+ For more details see <a href="../core-concepts#streams_processing_guarantee">Processing Guarantees</a>.
</div></blockquote>
</div>
<div class="section" id="replication-factor">
@@ -523,153 +729,79 @@
<span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="n">streamsConfig</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">ROCKSDB_CONFIG_SETTER_CLASS_CONFIG</span><span class="o">,</span> <span class="n">CustomRocksDBConfig</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
</pre></div>
- </div>
- <dl class="docutils">
- <dt>Notes for example:</dt>
- <dd><ol class="first last arabic simple">
- <li><code class="docutils literal"><span class="pre">BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();</span></code> Get a reference to the existing table config rather than create a new one, so you don't accidentally overwrite defaults such as the <code class="docutils literal"><span class="pre">BloomFilter</span></code>, which is an important optimization.
- <li><code class="docutils literal"><span class="pre">tableConfig.setBlockSize(16</span> <span class="pre">*</span> <span class="pre">1024L);</span></code> Modify the default <a class="reference external" href="https://github.com/apache/kafka/blob/2.3/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L79">block size</a> per these instructions from the <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/Memory- [...]
- <li><code class="docutils literal"><span class="pre">tableConfig.setCacheIndexAndFilterBlocks(true);</span></code> Do not let the index and filter blocks grow unbounded. For more information, see the <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/Block-Cache#caching-index-and-filter-blocks">RocksDB GitHub</a>.</li>
- <li><code class="docutils literal"><span class="pre">options.setMaxWriteBufferNumber(2);</span></code> See the advanced options in the <a class="reference external" href="https://github.com/facebook/rocksdb/blob/8dee8cad9ee6b70fd6e1a5989a8156650a70c04f/include/rocksdb/advanced_options.h#L103">RocksDB GitHub</a>.</li>
- <li><code class="docutils literal"><span class="pre">cache.close();</span></code> To avoid memory leaks, you must close any objects you constructed that extend org.rocksdb.RocksObject. See <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/RocksJava-Basics#memory-management">RocksJava docs</a> for more details.</li>
- </ol>
- </dd>
- </dl>
- </div></blockquote>
- </div>
- </div>
- </blockquote>
- </div>
- <div class="section" id="state-dir">
- <h4><a class="toc-backref" href="#id14">state.dir</a><a class="headerlink" href="#state-dir" title="Permalink to this headline"></a></h4>
- <blockquote>
- <div>The state directory. Kafka Streams persists local states under the state directory. Each application has a subdirectory on its hosting
- machine that is located under the state directory. The name of the subdirectory is the application ID. The state stores associated
- with the application are created under this subdirectory. When running multiple instances of the same application on a single machine,
- this path must be unique for each such instance.</div>
- </blockquote>
- </div>
- <div class="section" id="timestamp-extractor">
- <span id="streams-developer-guide-timestamp-extractor"></span><h4><a class="toc-backref" href="#id15">timestamp.extractor</a><a class="headerlink" href="#timestamp-extractor" title="Permalink to this headline"></a></h4>
- <blockquote>
- <div><p>A timestamp extractor pulls a timestamp from an instance of <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html">ConsumerRecord</a>.
- Timestamps are used to control the progress of streams.</p>
- <p>The default extractor is
- <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/FailOnInvalidTimestamp.html">FailOnInvalidTimestamp</a>.
- This extractor retrieves built-in timestamps that are automatically embedded into Kafka messages by the Kafka producer
- client since
- <a class="reference external" href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-32+-+Add+timestamps+to+Kafka+message">Kafka version 0.10</a>.
- Depending on the setting of Kafka’s server-side <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> broker and <code class="docutils literal"><span class="pre">message.timestamp.type</span></code> topic parameters,
- this extractor provides you with:</p>
- <ul class="simple">
- <li><strong>event-time</strong> processing semantics if <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> is set to <code class="docutils literal"><span class="pre">CreateTime</span></code> aka “producer time”
- (which is the default). This represents the time when a Kafka producer sent the original message. If you use Kafka’s
- official producer client, the timestamp represents milliseconds since the epoch.</li>
- <li><strong>ingestion-time</strong> processing semantics if <code class="docutils literal"><span class="pre">log.message.timestamp.type</span></code> is set to <code class="docutils literal"><span class="pre">LogAppendTime</span></code> aka “broker
- time”. This represents the time when the Kafka broker received the original message, in milliseconds since the epoch.</li>
- </ul>
- <p>The <code class="docutils literal"><span class="pre">FailOnInvalidTimestamp</span></code> extractor throws an exception if a record contains an invalid (i.e. negative) built-in
- timestamp, because Kafka Streams would not process this record but silently drop it. Invalid built-in timestamps can
- occur for various reasons: if for example, you consume a topic that is written to by pre-0.10 Kafka producer clients
- or by third-party producer clients that don’t support the new Kafka 0.10 message format yet; another situation where
- this may happen is after upgrading your Kafka cluster from <code class="docutils literal"><span class="pre">0.9</span></code> to <code class="docutils literal"><span class="pre">0.10</span></code>, where all the data that was generated
- with <code class="docutils literal"><span class="pre">0.9</span></code> does not include the <code class="docutils literal"><span class="pre">0.10</span></code> message timestamps.</p>
- <p>If you have data with invalid timestamps and want to process it, then there are two alternative extractors available.
- Both work on built-in timestamps, but handle invalid timestamps differently.</p>
- <ul class="simple">
- <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/LogAndSkipOnInvalidTimestamp.html">LogAndSkipOnInvalidTimestamp</a>:
- This extractor logs a warn message and returns the invalid timestamp to Kafka Streams, which will not process but
- silently drop the record.
- This log-and-skip strategy allows Kafka Streams to make progress instead of failing if there are records with an
- invalid built-in timestamp in your input data.</li>
- <li><a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/UsePartitionTimeOnInvalidTimestamp.html">UsePartitionTimeOnInvalidTimestamp</a>.
- This extractor returns the record’s built-in timestamp if it is valid (i.e. not negative). If the record does not
- have a valid built-in timestamps, the extractor returns the previously extracted valid timestamp from a record of the
- same topic partition as the current record as a timestamp estimation. In case that no timestamp can be estimated, it
- throws an exception.</li>
- </ul>
- <p>Another built-in extractor is
- <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/streams/processor/WallclockTimestampExtractor.html">WallclockTimestampExtractor</a>.
- This extractor does not actually “extract” a timestamp from the consumed record but rather returns the current time in
- milliseconds from the system clock (think: <code class="docutils literal"><span class="pre">System.currentTimeMillis()</span></code>), which effectively means Streams will operate
- on the basis of the so-called <strong>processing-time</strong> of events.</p>
- <p>You can also provide your own timestamp extractors, for instance to retrieve timestamps embedded in the payload of
- messages. If you cannot extract a valid timestamp, you can either throw an exception, return a negative timestamp, or
- estimate a timestamp. Returning a negative timestamp will result in data loss – the corresponding record will not be
- processed but silently dropped. If you want to estimate a new timestamp, you can use the value provided via
- <code class="docutils literal"><span class="pre">previousTimestamp</span></code> (i.e., a Kafka Streams timestamp estimation). Here is an example of a custom
- <code class="docutils literal"><span class="pre">TimestampExtractor</span></code> implementation:</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="kn">import</span> <span class="nn">org.apache.kafka.clients.consumer.ConsumerRecord</span><span class="o">;</span>
-<span class="kn">import</span> <span class="nn">org.apache.kafka.streams.processor.TimestampExtractor</span><span class="o">;</span>
-
-<span class="c1">// Extracts the embedded timestamp of a record (giving you "event-time" semantics).</span>
-<span class="kd">public</span> <span class="kd">class</span> <span class="nc">MyEventTimeExtractor</span> <span class="kd">implements</span> <span class="n">TimestampExtractor</span> <span class="o">{</span>
-
- <span class="nd">@Override</span>
- <span class="kd">public</span> <span class="kt">long</span> <span class="nf">extract</span><span class="o">(</span><span class="kd">final</span> <span class="n">ConsumerRecord</span><span class="o"><</span><span class="n">Object</span><span class="o">,</span> <span class="n">Object</span><span class="o">></span> <span class="n">record</span><span class="o">,</span> <span class="kd">final</span> <span class="kt">long</span> <span class="n">previousTimestamp</span><span class="o">) [...]
- <span class="c1">// `Foo` is your own custom class, which we assume has a method that returns</span>
- <span class="c1">// the embedded timestamp (milliseconds since midnight, January 1, 1970 UTC).</span>
- <span class="kt">long</span> <span class="n">timestamp</span> <span class="o">=</span> <span class="o">-</span><span class="mi">1</span><span class="o">;</span>
- <span class="kd">final</span> <span class="n">Foo</span> <span class="n">myPojo</span> <span class="o">=</span> <span class="o">(</span><span class="n">Foo</span><span class="o">)</span> <span class="n">record</span><span class="o">.</span><span class="na">value</span><span class="o">();</span>
- <span class="k">if</span> <span class="o">(</span><span class="n">myPojo</span> <span class="o">!=</span> <span class="kc">null</span><span class="o">)</span> <span class="o">{</span>
- <span class="n">timestamp</span> <span class="o">=</span> <span class="n">myPojo</span><span class="o">.</span><span class="na">getTimestampInMillis</span><span class="o">();</span>
- <span class="o">}</span>
- <span class="k">if</span> <span class="o">(</span><span class="n">timestamp</span> <span class="o"><</span> <span class="mi">0</span><span class="o">)</span> <span class="o">{</span>
- <span class="c1">// Invalid timestamp! Attempt to estimate a new timestamp,</span>
- <span class="c1">// otherwise fall back to wall-clock time (processing-time).</span>
- <span class="k">if</span> <span class="o">(</span><span class="n">previousTimestamp</span> <span class="o">>=</span> <span class="mi">0</span><span class="o">)</span> <span class="o">{</span>
- <span class="k">return</span> <span class="n">previousTimestamp</span><span class="o">;</span>
- <span class="o">}</span> <span class="k">else</span> <span class="o">{</span>
- <span class="k">return</span> <span class="n">System</span><span class="o">.</span><span class="na">currentTimeMillis</span><span class="o">();</span>
- <span class="o">}</span>
- <span class="o">}</span>
- <span class="o">}</span>
-
-<span class="o">}</span>
-</pre></div>
- </div>
- <p>You would then define the custom timestamp extractor in your Streams configuration as follows:</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="kn">import</span> <span class="nn">java.util.Properties</span><span class="o">;</span>
-<span class="kn">import</span> <span class="nn">org.apache.kafka.streams.StreamsConfig</span><span class="o">;</span>
-
-<span class="n">Properties</span> <span class="n">streamsConfiguration</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
-<span class="n">streamsConfiguration</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG</span><span class="o">,</span> <span class="n">MyEventTimeExtractor</span><span class="o">.</span><span class="na">class</span><span class="o">);</span>
-</pre></div>
</div>
+ <dl class="docutils">
+ <dt>Notes for example:</dt>
+ <dd><ol class="first last arabic simple">
+ <li><code class="docutils literal"><span class="pre">BlockBasedTableConfig tableConfig = (BlockBasedTableConfig) options.tableFormatConfig();</span></code> Get a reference to the existing table config rather than create a new one, so you don't accidentally overwrite defaults such as the <code class="docutils literal"><span class="pre">BloomFilter</span></code>, which is an important optimization.
+ <li><code class="docutils literal"><span class="pre">tableConfig.setBlockSize(16</span> <span class="pre">*</span> <span class="pre">1024L);</span></code> Modify the default <a class="reference external" href="https://github.com/apache/kafka/blob/2.3/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java#L79">block size</a> per these instructions from the <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/Memory-usage- [...]
+ <li><code class="docutils literal"><span class="pre">tableConfig.setCacheIndexAndFilterBlocks(true);</span></code> Do not let the index and filter blocks grow unbounded. For more information, see the <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/Block-Cache#caching-index-and-filter-blocks">RocksDB GitHub</a>.</li>
+ <li><code class="docutils literal"><span class="pre">options.setMaxWriteBufferNumber(2);</span></code> See the advanced options in the <a class="reference external" href="https://github.com/facebook/rocksdb/blob/8dee8cad9ee6b70fd6e1a5989a8156650a70c04f/include/rocksdb/advanced_options.h#L103">RocksDB GitHub</a>.</li>
+ <li><code class="docutils literal"><span class="pre">cache.close();</span></code> To avoid memory leaks, you must close any objects you constructed that extend org.rocksdb.RocksObject. See <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/RocksJava-Basics#memory-management">RocksJava docs</a> for more details.</li>
+ </ol>
+ </dd>
+ </dl>
</div></blockquote>
</div>
- <div class="section" id="upgrade-from">
- <h4><a class="toc-backref" href="#id14">upgrade.from</a><a class="headerlink" href="#upgrade-from" title="Permalink to this headline"></a></h4>
- <blockquote>
- <div>
- The version you are upgrading from. It is important to set this config when performing a rolling upgrade to certain versions, as described in the upgrade guide.
- You should set this config to the appropriate version before bouncing your instances and upgrading them to the newer version. Once everyone is on the
- newer version, you should remove this config and do a second rolling bounce. It is only necessary to set this config and follow the two-bounce upgrade path
- when upgrading from below version 2.0, or when upgrading to 2.4+ from any version lower than 2.4.
- </div>
- </blockquote>
- </div>
</div>
- <div class="section" id="kafka-consumers-and-producer-configuration-parameters">
- <h3><a class="toc-backref" href="#id16">Kafka consumers, producer and admin client configuration parameters</a><a class="headerlink" href="#kafka-consumers-and-producer-configuration-parameters" title="Permalink to this headline"></a></h3>
- <p>You can specify parameters for the Kafka <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/package-summary.html">consumers</a>, <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/producer/package-summary.html">producers</a>,
- and <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/kafka/clients/admin/package-summary.html">admin client</a> that are used internally.
- The consumer, producer and admin client settings are defined by specifying parameters in a <code class="docutils literal"><span class="pre">StreamsConfig</span></code> instance.</p>
- <p>In this example, the Kafka <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/ConsumerConfig.html#SESSION_TIMEOUT_MS_CONFIG">consumer session timeout</a> is configured to be 60000 milliseconds in the Streams settings:</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+ </blockquote>
+ </div>
+ <div class="section" id="state-dir">
+ <h4><a class="toc-backref" href="#id14">state.dir</a><a class="headerlink" href="#state-dir" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>The state directory. Kafka Streams persists local states under the state directory. Each application has a subdirectory on its hosting
+ machine that is located under the state directory. The name of the subdirectory is the application ID. The state stores associated
+ with the application are created under this subdirectory. When running multiple instances of the same application on a single machine,
+ this path must be unique for each such instance.</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>
+ <div>
+ <p>
+ You can tell Streams to apply topology optimizations by setting this config. The optimizations are currently all or none and disabled by default.
+ These optimizations include moving/reducing repartition topics and reusing the source topic as the changelog for source KTables. It is recommended to enable this.
+ </p>
+ <p>
+ Note that as of 2.3, you need to do two things to enable optimizations. In addition to setting this config to <code>StreamsConfig.OPTIMIZE</code>, you'll need to pass in your
+ configuration properties when building your topology by using the overloaded <code>StreamsBuilder.build(Properties)</code> method.
+ For example <code>KafkaStreams myStream = new KafkaStreams(streamsBuilder.build(properties), properties)</code>.
+ </p>
+ </div></blockquote>
+ </div>
+ <div class="section" id="upgrade-from">
+ <h4><a class="toc-backref" href="#id14">upgrade.from</a><a class="headerlink" href="#upgrade-from" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>
+ The version you are upgrading from. It is important to set this config when performing a rolling upgrade to certain versions, as described in the upgrade guide.
+ You should set this config to the appropriate version before bouncing your instances and upgrading them to the newer version. Once everyone is on the
+ newer version, you should remove this config and do a second rolling bounce. It is only necessary to set this config and follow the two-bounce upgrade path
+ when upgrading from below version 2.0, or when upgrading to 2.4+ from any version lower than 2.4.
+ </div>
+ </blockquote>
+ </div>
+ </div>
+ <div class="section" id="kafka-consumers-and-producer-configuration-parameters">
+ <h3><a class="toc-backref" href="#id16">Kafka consumers, producer and admin client configuration parameters</a><a class="headerlink" href="#kafka-consumers-and-producer-configuration-parameters" title="Permalink to this headline"></a></h3>
+ <p>You can specify parameters for the Kafka <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/package-summary.html">consumers</a>, <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/producer/package-summary.html">producers</a>,
+ and <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/kafka/clients/admin/package-summary.html">admin client</a> that are used internally.
+ The consumer, producer and admin client settings are defined by specifying parameters in a <code class="docutils literal"><span class="pre">StreamsConfig</span></code> instance.</p>
+ <p>In this example, the Kafka <a class="reference external" href="/{{version}}/javadoc/org/apache/kafka/clients/consumer/ConsumerConfig.html#SESSION_TIMEOUT_MS_CONFIG">consumer session timeout</a> is configured to be 60000 milliseconds in the Streams settings:</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="c1">// Example of a "normal" setting for Kafka Streams</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">BOOTSTRAP_SERVERS_CONFIG</span><span class="o">,</span> <span class="s">"kafka-broker-01:9092"</span><span class="o">);</span>
<span class="c1">// Customize the Kafka consumer settings of your Streams application</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">ConsumerConfig</span><span class="o">.</span><span class="na">SESSION_TIMEOUT_MS_CONFIG</span><span class="o">,</span> <span class="mi">60000</span><span class="o">);</span>
</pre></div>
- </div>
- <div class="section" id="naming">
- <h4><a class="toc-backref" href="#id17">Naming</a><a class="headerlink" href="#naming" title="Permalink to this headline"></a></h4>
- <p>Some consumer, producer and admin client configuration parameters use the same parameter name, and Kafka Streams library itself also uses some parameters that share the same name with its embedded client. For example, <code class="docutils literal"><span class="pre">send.buffer.bytes</span></code> and
- <code class="docutils literal"><span class="pre">receive.buffer.bytes</span></code> are used to configure TCP buffers; <code class="docutils literal"><span class="pre">request.timeout.ms</span></code> and <code class="docutils literal"><span class="pre">retry.backoff.ms</span></code> control retries for client request;
- <code class="docutils literal"><span class="pre">retries</span></code> are used to configure how many retries are allowed when handling retriable errors from broker request responses.
- You can avoid duplicate names by prefix parameter names with <code class="docutils literal"><span class="pre">consumer.</span></code>, <code class="docutils literal"><span class="pre">producer.</span></code>, or <code class="docutils literal"><span class="pre">admin.</span></code> (e.g., <code class="docutils literal"><span class="pre">consumer.send.buffer.bytes</span></code> and <code class="docutils literal"><span class="pre">producer.send.buffer.bytes</span></code>).</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+ </div>
+ <div class="section" id="naming">
+ <h4><a class="toc-backref" href="#id17">Naming</a><a class="headerlink" href="#naming" title="Permalink to this headline"></a></h4>
+ <p>Some consumer, producer and admin client configuration parameters use the same parameter name, and Kafka Streams library itself also uses some parameters that share the same name with its embedded client. For example, <code class="docutils literal"><span class="pre">send.buffer.bytes</span></code> and
+ <code class="docutils literal"><span class="pre">receive.buffer.bytes</span></code> are used to configure TCP buffers; <code class="docutils literal"><span class="pre">request.timeout.ms</span></code> and <code class="docutils literal"><span class="pre">retry.backoff.ms</span></code> control retries for client request;
+ <code class="docutils literal"><span class="pre">retries</span></code> are used to configure how many retries are allowed when handling retriable errors from broker request responses.
+ You can avoid duplicate names by prefix parameter names with <code class="docutils literal"><span class="pre">consumer.</span></code>, <code class="docutils literal"><span class="pre">producer.</span></code>, or <code class="docutils literal"><span class="pre">admin.</span></code> (e.g., <code class="docutils literal"><span class="pre">consumer.send.buffer.bytes</span></code> and <code class="docutils literal"><span class="pre">producer.send.buffer.bytes</span></code>).</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="c1">// same value for consumer, producer, and admin client</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"PARAMETER_NAME"</span><span class="o">,</span> <span class="s">"value"</span><span class="o">);</span>
<span class="c1">// different values for consumer and producer</span>
@@ -681,14 +813,14 @@
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">producerPrefix</span><span class="o">(</span><span class="s">"PARAMETER_NAME"</span><span class="o">),</span> <span class="s">"producer-value"</span><span class="o">);</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">adminClientPrefix</span><span class="o">(</span><span class="s">"PARAMETER_NAME"</span><span class="o">),</span> <span class="s">"admin-value"</span><span class="o">);</span>
</pre></div>
- <p>You could further separate consumer configuration by adding different prefixes:</p>
- <ul class="simple">
- <li><code class="docutils literal"><span class="pre">main.consumer.</span></code> for main consumer which is the default consumer of stream source.</li>
- <li><code class="docutils literal"><span class="pre">restore.consumer.</span></code> for restore consumer which is in charge of state store recovery.</li>
- <li><code class="docutils literal"><span class="pre">global.consumer.</span></code> for global consumer which is used in global KTable construction.</li>
- </ul>
- <p>For example, if you only want to set restore consumer config without touching other consumers' settings, you could simply use <code class="docutils literal"><span class="pre">restore.consumer.</span></code> to set the config.</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+ <p>You could further separate consumer configuration by adding different prefixes:</p>
+ <ul class="simple">
+ <li><code class="docutils literal"><span class="pre">main.consumer.</span></code> for main consumer which is the default consumer of stream source.</li>
+ <li><code class="docutils literal"><span class="pre">restore.consumer.</span></code> for restore consumer which is in charge of state store recovery.</li>
+ <li><code class="docutils literal"><span class="pre">global.consumer.</span></code> for global consumer which is used in global KTable construction.</li>
+ </ul>
+ <p>For example, if you only want to set restore consumer config without touching other consumers' settings, you could simply use <code class="docutils literal"><span class="pre">restore.consumer.</span></code> to set the config.</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="c1">// same config value for all consumer types</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"consumer.PARAMETER_NAME"</span><span class="o">,</span> <span class="s">"general-consumer-value"</span><span class="o">);</span>
<span class="c1">// set a different restore consumer config. This would make restore consumer take restore-consumer-value,</span>
@@ -697,103 +829,103 @@
<span class="c1">// alternatively, you can use</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">restoreConsumerPrefix</span><span class="o">(</span><span class="s">"PARAMETER_NAME"</span><span class="o">),</span> <span class="s">"restore-consumer-value"</span><span class="o">);</span>
</pre></div>
- </div>
- <p> Same applied to <code class="docutils literal"><span class="pre">main.consumer.</span></code> and <code class="docutils literal"><span class="pre">main.consumer.</span></code>, if you only want to specify one consumer type config.</p>
- <p> Additionally, to configure the internal repartition/changelog topics, you could use the <code class="docutils literal"><span class="pre">topic.</span></code> prefix, followed by any of the standard topic configs.</p>
- <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
+ </div>
+ <p> Same applied to <code class="docutils literal"><span class="pre">main.consumer.</span></code> and <code class="docutils literal"><span class="pre">main.consumer.</span></code>, if you only want to specify one consumer type config.</p>
+ <p> Additionally, to configure the internal repartition/changelog topics, you could use the <code class="docutils literal"><span class="pre">topic.</span></code> prefix, followed by any of the standard topic configs.</p>
+ <div class="highlight-java"><div class="highlight"><pre><span></span><span class="n">Properties</span> <span class="n">streamsSettings</span> <span class="o">=</span> <span class="k">new</span> <span class="n">Properties</span><span class="o">();</span>
<span class="c1">// Override default for both changelog and repartition topics</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="s">"topic.PARAMETER_NAME"</span><span class="o">,</span> <span class="s">"topic-value"</span><span class="o">);</span>
<span class="c1">// alternatively, you can use</span>
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">topicPrefix</span><span class="o">(</span><span class="s">"PARAMETER_NAME"</span><span class="o">),</span> <span class="s">"topic-value"</span><span class="o">);</span>
</pre></div>
- </div>
- </div>
- </div>
- <div class="section" id="default-values">
- <h4><a class="toc-backref" href="#id18">Default Values</a><a class="headerlink" href="#default-values" title="Permalink to this headline"></a></h4>
- <p>Kafka Streams uses different default values for some of the underlying client configs, which are summarized below. For detailed descriptions
- of these configs, see <a class="reference external" href="http://kafka.apache.org/0100/documentation.html#producerconfigs">Producer Configs</a>
- and <a class="reference external" href="http://kafka.apache.org/0100/documentation.html#newconsumerconfigs">Consumer Configs</a>.</p>
- <table border="1" class="non-scrolling-table docutils">
- <thead valign="bottom">
- <tr class="row-odd"><th class="head">Parameter Name</th>
- <th class="head">Corresponding Client</th>
- <th class="head">Streams Default</th>
- </tr>
- </thead>
- <tbody valign="top">
- <tr class="row-even"><td>auto.offset.reset</td>
- <td>Consumer</td>
- <td>earliest</td>
- </tr>
- <tr class="row-even"><td>linger.ms</td>
- <td>Producer</td>
- <td>100</td>
- </tr>
- <tr class="row-odd"><td>max.poll.interval.ms</td>
- <td>Consumer</td>
- <td>Integer.MAX_VALUE</td>
- </tr>
- <tr class="row-even"><td>max.poll.records</td>
- <td>Consumer</td>
- <td>1000</td>
- </tr>
- </tbody>
- </table>
- </div>
- <div class="section" id="parameters-controlled-by-kafka-streams">
- <h3><a class="toc-backref" href="#id26">Parameters controlled by Kafka Streams</a><a class="headerlink" href="#parameters-controlled-by-kafka-streams" title="Permalink to this headline"></a></h3>
- <p>Kafka Streams assigns the following configuration parameters. If you try to change
- <code class="docutils literal"><span class="pre">allow.auto.create.topics</span></code>, your value
- is ignored and setting it has no effect in a Kafka Streams application. You can set the other parameters.
- Kafka Streams sets them to different default values than a plain
- <code class="docutils literal"><span class="pre">KafkaConsumer</span></code>.
- <p>Kafka Streams uses the <code class="docutils literal"><span class="pre">client.id</span></code>
- parameter to compute derived client IDs for internal clients. If you don't set
- <code class="docutils literal"><span class="pre">client.id</span></code>, Kafka Streams sets it to
- <code class="docutils literal"><span class="pre"><application.id>-<random-UUID></span></code>.
- <table border="1" class="non-scrolling-table docutils">
- <colgroup>
- <col width="50%">
- <col width="19%">
- <col width="31%">
- </colgroup>
- <thead valign="bottom">
- <tr class="row-odd"><th class="head">Parameter Name</th>
- <th class="head">Corresponding Client</th>
- <th class="head">Streams Default</th>
- </tr>
- </thead>
- <tbody valign="top">
- <tr class="row-odd"><td>allow.auto.create.topics</td>
- <td>Consumer</td>
- <td>false</td>
- </tr>
- <tr class="row-even"><td>auto.offset.reset</td>
- <td>Consumer</td>
- <td>earliest</td>
- </tr>
- <tr class="row-odd"><td>linger.ms</td>
- <td>Producer</td>
- <td>100</td>
- </tr>
- <tr class="row-even"><td>max.poll.interval.ms</td>
- <td>Consumer</td>
- <td>300000</td>
- </tr>
- <tr class="row-odd"><td>max.poll.records</td>
- <td>Consumer</td>
- <td>1000</td>
- </tr>
- </tbody>
- </table>
- <div class="section" id="enable-auto-commit">
- <span id="streams-developer-guide-consumer-auto-commit"></span><h4><a class="toc-backref" href="#id19">enable.auto.commit</a><a class="headerlink" href="#enable-auto-commit" title="Permalink to this headline"></a></h4>
- <blockquote>
- <div>The consumer auto commit. To guarantee at-least-once processing semantics and turn off auto commits, Kafka Streams overrides this consumer config
- value to <code class="docutils literal"><span class="pre">false</span></code>. Consumers will only commit explicitly via <em>commitSync</em> calls when the Kafka Streams library or a user decides
- to commit the current processing state.</div></blockquote>
</div>
+ </div>
+ </div>
+ <div class="section" id="default-values">
+ <h4><a class="toc-backref" href="#id18">Default Values</a><a class="headerlink" href="#default-values" title="Permalink to this headline"></a></h4>
+ <p>Kafka Streams uses different default values for some of the underlying client configs, which are summarized below. For detailed descriptions
+ of these configs, see <a class="reference external" href="http://kafka.apache.org/0100/documentation.html#producerconfigs">Producer Configs</a>
+ and <a class="reference external" href="http://kafka.apache.org/0100/documentation.html#newconsumerconfigs">Consumer Configs</a>.</p>
+ <table border="1" class="non-scrolling-table docutils">
+ <thead valign="bottom">
+ <tr class="row-odd"><th class="head">Parameter Name</th>
+ <th class="head">Corresponding Client</th>
+ <th class="head">Streams Default</th>
+ </tr>
+ </thead>
+ <tbody valign="top">
+ <tr class="row-even"><td>auto.offset.reset</td>
+ <td>Consumer</td>
+ <td>earliest</td>
+ </tr>
+ <tr class="row-even"><td>linger.ms</td>
+ <td>Producer</td>
+ <td>100</td>
+ </tr>
+ <tr class="row-odd"><td>max.poll.interval.ms</td>
+ <td>Consumer</td>
+ <td>Integer.MAX_VALUE</td>
+ </tr>
+ <tr class="row-even"><td>max.poll.records</td>
+ <td>Consumer</td>
+ <td>1000</td>
+ </tr>
+ </tbody>
+ </table>
+ </div>
+ <div class="section" id="parameters-controlled-by-kafka-streams">
+ <h3><a class="toc-backref" href="#id26">Parameters controlled by Kafka Streams</a><a class="headerlink" href="#parameters-controlled-by-kafka-streams" title="Permalink to this headline"></a></h3>
+ <p>Kafka Streams assigns the following configuration parameters. If you try to change
+ <code class="docutils literal"><span class="pre">allow.auto.create.topics</span></code>, your value
+ is ignored and setting it has no effect in a Kafka Streams application. You can set the other parameters.
+ Kafka Streams sets them to different default values than a plain
+ <code class="docutils literal"><span class="pre">KafkaConsumer</span></code>.
+ <p>Kafka Streams uses the <code class="docutils literal"><span class="pre">client.id</span></code>
+ parameter to compute derived client IDs for internal clients. If you don't set
+ <code class="docutils literal"><span class="pre">client.id</span></code>, Kafka Streams sets it to
+ <code class="docutils literal"><span class="pre"><application.id>-<random-UUID></span></code>.
+ <table border="1" class="non-scrolling-table docutils">
+ <colgroup>
+ <col width="50%">
+ <col width="19%">
+ <col width="31%">
+ </colgroup>
+ <thead valign="bottom">
+ <tr class="row-odd"><th class="head">Parameter Name</th>
+ <th class="head">Corresponding Client</th>
+ <th class="head">Streams Default</th>
+ </tr>
+ </thead>
+ <tbody valign="top">
+ <tr class="row-odd"><td>allow.auto.create.topics</td>
+ <td>Consumer</td>
+ <td>false</td>
+ </tr>
+ <tr class="row-even"><td>auto.offset.reset</td>
+ <td>Consumer</td>
+ <td>earliest</td>
+ </tr>
+ <tr class="row-odd"><td>linger.ms</td>
+ <td>Producer</td>
+ <td>100</td>
+ </tr>
+ <tr class="row-even"><td>max.poll.interval.ms</td>
+ <td>Consumer</td>
+ <td>300000</td>
+ </tr>
+ <tr class="row-odd"><td>max.poll.records</td>
+ <td>Consumer</td>
+ <td>1000</td>
+ </tr>
+ </tbody>
+ </table>
+ <div class="section" id="enable-auto-commit">
+ <span id="streams-developer-guide-consumer-auto-commit"></span><h4><a class="toc-backref" href="#id19">enable.auto.commit</a><a class="headerlink" href="#enable-auto-commit" title="Permalink to this headline"></a></h4>
+ <blockquote>
+ <div>The consumer auto commit. To guarantee at-least-once processing semantics and turn off auto commits, Kafka Streams overrides this consumer config
+ value to <code class="docutils literal"><span class="pre">false</span></code>. Consumers will only commit explicitly via <em>commitSync</em> calls when the Kafka Streams library or a user decides
+ to commit the current processing state.</div></blockquote>
+ </div>
<div class="section" id="recommended-configuration-parameters-for-resiliency">
<h3><a class="toc-backref" href="#id21">Recommended configuration parameters for resiliency</a><a class="headerlink" href="#recommended-configuration-parameters-for-resiliency" title="Permalink to this headline"></a></h3>
<p>There are several Kafka and Kafka Streams configuration options that need to be configured explicitly for resiliency in face of broker failures:</p>
@@ -849,36 +981,36 @@
<span class="n">streamsSettings</span><span class="o">.</span><span class="na">put</span><span class="o">(</span><span class="n">StreamsConfig</span><span class="o">.</span><span class="na">producerPrefix</span><span class="o">(</span><span class="n">ProducerConfig</span><span class="o">.</span><span class="na">ACKS_CONFIG</span><span class="o">),</span> <span class="s">"all"</span><span class="o">);</span>
</pre></div>
</div>
-</div>
-</div>
-</div>
-</div>
+ </div>
+ </div>
+ </div>
+ </div>
- </div>
- </div>
- <div class="pagination">
- <a href="/{{version}}/documentation/streams/developer-guide/write-streams" class="pagination__btn pagination__btn__prev">Previous</a>
- <a href="/{{version}}/documentation/streams/developer-guide/dsl-api" class="pagination__btn pagination__btn__next">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>
- <li><a href="/documentation/streams/developer-guide/">Developer Guide</a></li>
- </ul>
- <div class="p-content"></div>
- </div>
- </div>
- <!--#include virtual="../../../includes/_footer.htm" -->
- <script>
+ </div>
+ </div>
+ <div class="pagination">
+ <a href="/{{version}}/documentation/streams/developer-guide/write-streams" class="pagination__btn pagination__btn__prev">Previous</a>
+ <a href="/{{version}}/documentation/streams/developer-guide/dsl-api" class="pagination__btn pagination__btn__next">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>
+ <li><a href="/documentation/streams/developer-guide/">Developer Guide</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');
diff --git a/26/streams/developer-guide/memory-mgmt.html b/26/streams/developer-guide/memory-mgmt.html
index 5f1158f..3afa3aa 100644
--- a/26/streams/developer-guide/memory-mgmt.html
+++ b/26/streams/developer-guide/memory-mgmt.html
@@ -206,7 +206,10 @@
<span class="o">}</span>
<span class="o">}</span>
</div>
- <sup id="fn1">1. INDEX_FILTER_BLOCK_RATIO can be used to set a fraction of the block cache to set aside for "high priority" (aka index and filter) blocks, preventing them from being evicted by data blocks. See the full signature of the <a class="reference external" href="https://github.com/facebook/rocksdb/blob/master/java/src/main/java/org/rocksdb/LRUCache.java#L72">LRUCache constructor</a>. </sup>
+ <sup id="fn1">1. INDEX_FILTER_BLOCK_RATIO can be used to set a fraction of the block cache to set aside for "high priority" (aka index and filter) blocks, preventing them from being evicted by data blocks. See the full signature of the <a class="reference external" href="https://github.com/facebook/rocksdb/blob/master/java/src/main/java/org/rocksdb/LRUCache.java#L72">LRUCache constructor</a>.
+ NOTE: the boolean parameter in the cache constructor lets you control whether the cache should enforce a strict memory limit by failing the read or iteration in the rare cases where it might go larger than its capacity. Due to a
+ <a class="reference external" href="https://github.com/facebook/rocksdb/issues/6247">bug in RocksDB</a>, this option cannot be used
+ if the write buffer memory is also counted against the cache. If you set this to true, you should NOT pass the cache in to the <code>WriteBufferManager</code> and just control the write buffer and cache memory separately.</sup>
<br>
<sup id="fn2">2. This must be set in order for INDEX_FILTER_BLOCK_RATIO to take effect (see footnote 1) as described in the <a class="reference external" href="https://github.com/facebook/rocksdb/wiki/Block-Cache#caching-index-and-filter-blocks">RocksDB docs</a></sup>
<br>
diff --git a/26/streams/developer-guide/running-app.html b/26/streams/developer-guide/running-app.html
index b2c4fc5..991f6b4 100644
--- a/26/streams/developer-guide/running-app.html
+++ b/26/streams/developer-guide/running-app.html
@@ -110,6 +110,18 @@ $ java -cp path-to-app-fatjar.jar com.example.MyStreamsApp
<li>If a local state store exists, the changelog is replayed from the previously checkpointed offset. The changes are applied and the state is restored to the most recent snapshot. This method takes less time because it is applying a smaller portion of the changelog.</li>
</ul>
<p>For more information, see <a class="reference internal" href="config-streams.html#num-standby-replicas"><span class="std std-ref">Standby Replicas</span></a>.</p>
+ <p>
+ As of version 2.6, Streams will now do most of a task's restoration in the background through warmup replicas. These will be assigned to instances that need to restore a lot of state for a task.
+ A stateful active task will only be assigned to an instance once its state is within the configured
+ <a class="reference internal" href="config-streams.html#acceptable-recovery-lag"><span class="std std-ref"><code>acceptable.recovery.lag</code></span></a>, if one exists. This means that
+ most of the time, a task migration will <b>not</b> result in downtime for that task. It will remain active on the instance that's already caught up, while the instance that it's being
+ migrated to works on restoring the state. Streams will <a class="reference internal" href="config-streams.html#probing-rebalance-interval-ms"><span class="std std-ref">regularly probe</span></a> for warmup tasks that have finished restoring and transition them to active tasks when ready.
+ </p>
+ <p>
+ Note, the one exception to this task availability is if none of the instances have a caught up version of that task. In that case, we have no choice but to assign the active
+ task to an instance that is not caught up and will have to block further processing on restoration of the task's state from the changelog. If high availability is important
+ for your application, you are highly recommended to enable standbys.
+ </p>
</div>
<div class="section" id="determining-how-many-application-instances-to-run">
<h3><a class="toc-backref" href="#id8">Determining how many application instances to run</a><a class="headerlink" href="#determining-how-many-application-instances-to-run" title="Permalink to this headline"></a></h3>
diff --git a/26/streams/upgrade-guide.html b/26/streams/upgrade-guide.html
index b1af752..9d5a20e 100644
--- a/26/streams/upgrade-guide.html
+++ b/26/streams/upgrade-guide.html
@@ -42,7 +42,7 @@
<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> prepare your newly deployed {{fullDotVersion}} application instances for a second round of rolling bounces; make sure to remove the value for config <code>upgrade.from</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>
@@ -95,7 +95,19 @@
Note that you need brokers with version 2.5 or newer to use this feature.
</p>
<p>
- As of 2.6.0 Kafka Streams deprecates <code>KStream.through()<code> if favor of the new <code>KStream.repartition()</code> operator
+ For more highly available stateful applications, we've modified the task assignment algorithm to delay the movement of stateful active tasks to instances
+ that aren't yet caught up with that task's state. Instead, to migrate a task from one instance to another (eg when scaling out),
+ Streams will assign a warmup replica to the target instance so it can begin restoring the state while the active task stays available on an instance
+ that already had the task. The instances warming up tasks will communicate their progress to the group so that, once ready, Streams can move active
+ tasks to their new owners in the background. Check out <a href="https://cwiki.apache.org/confluence/x/0i4lBg">KIP-441</a>
+ for full details, including several new configs for control over this new feature.
+ </p>
+ <p>
+ New end-to-end latency metrics have been added. These task-level metrics will be logged at the INFO level and report the min and max end-to-end latency of a record at the beginning/source node(s)
+ and end/terminal node(s) of a task. See <a href="https://cwiki.apache.org/confluence/x/gBkRCQ">KIP-613</a> for more information.
+ </p>
+ <p>
+ As of 2.6.0 Kafka Streams deprecates <code>KStream.through()</code> if favor of the new <code>KStream.repartition()</code> operator
(as per <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-221%3A+Enhance+DSL+with+Connecting+Topic+Creation+and+Repartition+Hint">KIP-221</a>).
<code>KStream.repartition()</code> is similar to <code>KStream.through()</code>, however Kafka Streams will manage the topic for you.
If you need to write into and read back from a topic that you mange, you can fall back to use <code>KStream.to()</code> in combination with <code>StreamsBuilder#stream()</code>.