This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 23927a2037f MINOR: Fix broken relative links in docs/ (#22813)
23927a2037f is described below
commit 23927a2037f201536a7cc44313b2cbfaa5e543cf
Author: Ming-Yen Chung <[email protected]>
AuthorDate: Fri Jul 17 04:58:22 2026 +0800
MINOR: Fix broken relative links in docs/ (#22813)
Several docs/ cross-links use relative paths assuming a flat
single-page-HTML layout, pre-dating the move to per-page Hugo
directories (foo.md -> /foo/) on kafka.apache.org, so they resolve one
level too shallow and 404 — including the Kafka Connect quickstart link
and the new Streams/Connect security-model cross-links.
Reviewers: Chia-Ping Tsai <[email protected]>
---
docs/getting-started/upgrade.md | 2 +-
docs/kafka-connect/user-guide.md | 2 +-
docs/security/security-model-connect.md | 12 ++---
docs/security/security-model-streams.md | 4 +-
docs/security/security-model.md | 6 +--
docs/streams/developer-guide/app-reset-tool.md | 2 +-
docs/streams/developer-guide/config-streams.md | 14 +++---
docs/streams/developer-guide/datatypes.md | 6 +--
docs/streams/developer-guide/dsl-api.md | 56 +++++++++++-----------
.../streams/developer-guide/interactive-queries.md | 6 +--
docs/streams/developer-guide/manage-topics.md | 4 +-
docs/streams/developer-guide/memory-mgmt.md | 12 ++---
docs/streams/developer-guide/processor-api.md | 22 ++++-----
docs/streams/developer-guide/running-app.md | 12 ++---
docs/streams/developer-guide/security.md | 2 +-
docs/streams/developer-guide/testing.md | 4 +-
docs/streams/developer-guide/write-streams-app.md | 16 +++----
17 files changed, 91 insertions(+), 91 deletions(-)
diff --git a/docs/getting-started/upgrade.md b/docs/getting-started/upgrade.md
index ca6ab9923db..81b81e57c34 100644
--- a/docs/getting-started/upgrade.md
+++ b/docs/getting-started/upgrade.md
@@ -153,7 +153,7 @@ For further details, please refer to
[KIP-1120](https://cwiki.apache.org/conflue
* The metrics
`org.apache.kafka.server:type=AssignmentsManager.QueuedReplicaToDirAssignments`,
`org.apache.kafka.storage.internals.log:type=RemoteStorageThreadPool.RemoteLogReaderTaskQueueSize`,
and
`org.apache.kafka.storage.internals.log:type=RemoteStorageThreadPool.RemoteLogReaderAvgIdlePercent`
have been deprecated and will be removed in Kafka 5.0. As replacements, the
following metrics have been introduced, which report the same information:
`kafka.server:type=AssignmentsManager.Qu [...]
* A new metric `AvgIdleRatio` has been added to the `ControllerEventManager`
and `MetadataLoader` groups. These metrics measure the average idle ratio of
their respective event queue threads, providing visibility into how much time
each component spends waiting for events versus processing them. The metric
value ranges from 0.0 (always busy) to 1.0 (always idle).
* Deprecated `org.apache.kafka.streams.KafkaStreams$CloseOptions` and its
related methods, such as
`KafkaStreams#close(org.apache.kafka.streams.KafkaStreams$CloseOptions)`. As a
replacement, please use `org.apache.kafka.streams.CloseOptions` and
`KafkaStreams#close(org.apache.kafka.streams.CloseOptions)`. For further
details, please refer to
[KIP-1153](https://cwiki.apache.org/confluence/x/QAq9F).
- * A new implementation of `ConnectorClientConfigOverridePolicy`,
`AllowlistConnectorClientConfigOverridePolicy`, has been added. This enables
specifying the configurations that connectors can override via
`connector.client.config.override.allowlist`. From Kafka 5.0.0, this will be
the default
[connector.client.config.override.policy](documentation/#connectconfigs_connector.client.config.override.policy)
policy. The `PrincipalConnectorClientConfigOverridePolicy` policy is now
deprecated [...]
+ * A new implementation of `ConnectorClientConfigOverridePolicy`,
`AllowlistConnectorClientConfigOverridePolicy`, has been added. This enables
specifying the configurations that connectors can override via
`connector.client.config.override.allowlist`. From Kafka 5.0.0, this will be
the default
[connector.client.config.override.policy](https://kafka.apache.org/documentation/#connectconfigs_connector.client.config.override.policy)
policy. The `PrincipalConnectorClientConfigOverridePolicy` [...]
* It is now possible to specify the start time for a Kafka Streams
punctuation, instead of relying on the non-deterministic time when you register
it. For further details, please refer to
[KIP-1146](https://cwiki.apache.org/confluence/x/9QqWF).
* Added an optional `--node-id` flag to the `FeatureCommand` command. It
specifies the node to describe. If not provided, an arbitrary node is used.
diff --git a/docs/kafka-connect/user-guide.md b/docs/kafka-connect/user-guide.md
index 3f46d54b0b1..8436962716b 100644
--- a/docs/kafka-connect/user-guide.md
+++ b/docs/kafka-connect/user-guide.md
@@ -26,7 +26,7 @@ type: docs
-->
-The [quickstart](../getting-started/quickstart) provides a brief example of
how to run a standalone version of Kafka Connect. This section describes how to
configure, run, and manage Kafka Connect in more detail.
+The [quickstart](../../getting-started/quickstart) provides a brief example of
how to run a standalone version of Kafka Connect. This section describes how to
configure, run, and manage Kafka Connect in more detail.
## Running Kafka Connect
diff --git a/docs/security/security-model-connect.md
b/docs/security/security-model-connect.md
index 613d7535cfe..2ebfda95383 100644
--- a/docs/security/security-model-connect.md
+++ b/docs/security/security-model-connect.md
@@ -26,11 +26,11 @@ type: docs
-->
-This page extends the [Apache Kafka security model](security-model) to Kafka
Connect. A worker authenticates to the Kafka cluster over a configured
`SASL_SSL`/`SSL` listener exactly like any other client, so everything the core
model says about authentication, authorization, and transport encryption to the
brokers applies unchanged. What follows covers only what Connect adds on top —
chiefly its own control plane, the REST API, and the fact that it runs
user-supplied code.
+This page extends the [Apache Kafka security model](../security-model) to
Kafka Connect. A worker authenticates to the Kafka cluster over a configured
`SASL_SSL`/`SSL` listener exactly like any other client, so everything the core
model says about authentication, authorization, and transport encryption to the
brokers applies unchanged. What follows covers only what Connect adds on top —
chiefly its own control plane, the REST API, and the fact that it runs
user-supplied code.
## Things You Need To Know
-- **Connect inherits the broker's client security model.** Authentication to
the brokers, broker-side authorization, and transport encryption are exactly as
described in the [core security model](security-model). This page only
describes what Connect layers on top.
+- **Connect inherits the broker's client security model.** Authentication to
the brokers, broker-side authorization, and transport encryption are exactly as
described in the [core security model](../security-model). This page only
describes what Connect layers on top.
- **The REST API is unauthenticated by default.** Out of the box, anyone who
can reach the REST port can create, reconfigure, stop, or delete any connector.
Because connectors and plugins run arbitrary code, REST access lets a caller
run anything the worker's installed plugins allow.
- **Connect plugins run arbitrary code.** Connectors, converters,
transformations, predicates, and REST extensions loaded from `plugin.path`
execute in the worker JVM with its privileges. Install only plugins you trust.
- **The REST API is a shared control plane with no per-connector isolation.**
There is no notion of connector ownership: any caller allowed onto the API can
act on every connector and read its configuration.
@@ -54,7 +54,7 @@ Connect enables no authentication on the REST API by default.
There are two comm
- **Reverse proxy.** Terminate authentication (mTLS, OIDC, basic auth, etc.)
in a proxy in front of the workers and allow only the proxy to reach the REST
port.
- **REST extension.** Register an authentication extension via
`rest.extension.classes`. The built-in `BasicAuthSecurityRestExtension`
performs JAAS-based HTTP basic authentication against a configured
`LoginModule`. The reference `PropertyFileLoginModule` is **not** intended for
production, as it stores credentials in cleartext; production deployments
should configure a `LoginModule` that authenticates against a real credential
store.
-REST authentication only establishes *who is calling*; it does not, on its
own, authorize that caller on a per-connector basis (see Authorization below).
Separately, the worker's authentication *to the Kafka brokers* uses the
standard `ssl.*`/`sasl.*` client configs described in the [core
model](security-model).
+REST authentication only establishes *who is calling*; it does not, on its
own, authorize that caller on a per-connector basis (see Authorization below).
Separately, the worker's authentication *to the Kafka brokers* uses the
standard `ssl.*`/`sasl.*` client configs described in the [core
model](../security-model).
## Authorization
@@ -70,12 +70,12 @@ The practical consequences are that you cannot grant
Connect REST API access to
Two independent channels need TLS:
-- **Worker-to-Kafka.** Configured with the standard `ssl.*` client properties,
exactly as in the [core model](security-model).
+- **Worker-to-Kafka.** Configured with the standard `ssl.*` client properties,
exactly as in the [core model](../security-model).
- **REST API (and inter-worker).** Enable an `https` listener. By default the
REST server reuses the worker's `ssl.*` settings; to configure the REST
endpoint independently of the Kafka client, use the `listeners.https.*`
prefixed properties (when the prefix is used, the unprefixed `ssl.*` settings
are ignored for the REST server). The same settings secure inter-worker
forwarding in distributed mode.
## Secrets in Configuration
-Connector configurations frequently contain credentials for external systems.
As in the [core model](security-model), reference them indirectly through a
`ConfigProvider` rather than inlining them, and set `allowed.paths` on the
file-based providers to constrain which directories they can read. Two
Connect-specific caveats:
+Connector configurations frequently contain credentials for external systems.
As in the [core model](../security-model), reference them indirectly through a
`ConfigProvider` rather than inlining them, and set `allowed.paths` on the
file-based providers to constrain which directories they can read. Two
Connect-specific caveats:
- **Config providers are resolved through the shared REST API.** A caller who
can guess or enumerate a provider alias can resolve its full value, so a
provider is only as isolated as the REST API in front of it.
- **What goes in the config topic depends on how secrets are supplied.** A
secret placed directly in a connector configuration is written to
`config.storage.topic` as an ordinary Kafka record, and is therefore only as
protected as that topic's ACLs and the brokers' at-rest story. A
`ConfigProvider` reference keeps the secret itself out of the topic — only the
template string (`${alias:fields}`) is stored, not the resolved value. Note,
however, that this does not hide the secret from REST [...]
@@ -86,7 +86,7 @@ Connect loads connectors, converters, single-message
transforms, predicates, `Co
## Known Non-Findings
-In line with the [core model's classification](security-model), the following
follow from Connect's design and are not, on their own, security
vulnerabilities:
+In line with the [core model's classification](../security-model), the
following follow from Connect's design and are not, on their own, security
vulnerabilities:
- **File-based connectors granting disk access.** Adding the file connectors
to a worker effectively grants read/write access to the worker's local disk.
This is the connector's intended function, not a flaw.
- **Local-disk-only weaknesses in the file-based config providers.** Issues
that require an attacker to already have local disk access on a Connect worker
— for example the ability to create arbitrary files or symlinks — are not
security issues, because such access is outside Connect's trust boundary.
Path-validation robustness in these providers may still be hardened
independently.
diff --git a/docs/security/security-model-streams.md
b/docs/security/security-model-streams.md
index 8508ce5c9a1..7904c2368f5 100644
--- a/docs/security/security-model-streams.md
+++ b/docs/security/security-model-streams.md
@@ -26,7 +26,7 @@ type: docs
-->
-This page extends the [Apache Kafka security model](security-model) to Kafka
Streams. From the cluster's perspective a Streams application is an ordinary
Kafka client and inherits the broker's authentication, authorization, and
transport-encryption model through the same `ssl.*` and `sasl.*` client configs.
+This page extends the [Apache Kafka security model](../security-model) to
Kafka Streams. From the cluster's perspective a Streams application is an
ordinary Kafka client and inherits the broker's authentication, authorization,
and transport-encryption model through the same `ssl.*` and `sasl.*` client
configs.
## Things You Need To Know
@@ -36,7 +36,7 @@ This page extends the [Apache Kafka security
model](security-model) to Kafka Str
## Known Non-Findings
-In line with the [core model's classification](security-model), the following
are not, on their own, security vulnerabilities:
+In line with the [core model's classification](../security-model), the
following are not, on their own, security vulnerabilities:
- **Application-level issues in a Streams topology.** Streams runs inside the
user's application, so bugs in user-supplied processors, state handling, or
topology code fall within the application's trust boundary, not the broker's.
- **Local state being readable on the host.** State stores and changelogs on
disk are protected by the host filesystem, not by Kafka; reading them requires
local access that is outside the model.
diff --git a/docs/security/security-model.md b/docs/security/security-model.md
index b7f96add11a..9b9d7929d7a 100644
--- a/docs/security/security-model.md
+++ b/docs/security/security-model.md
@@ -128,8 +128,8 @@ Broker, client, and Connect properties files contain
keystore passwords, SASL cr
The components built on top of the Kafka clients have their own security
models, covered on separate pages:
-- [Kafka Connect](security-model-connect)
-- [Kafka Streams](security-model-streams)
+- [Kafka Connect](../security-model-connect)
+- [Kafka Streams](../security-model-streams)
## Development and Test Tooling
@@ -162,7 +162,7 @@ The following follow directly from the model above and are
not, on their own, se
- **Unauthenticated or unencrypted access to a default cluster.** Security is
off by default; an open `PLAINTEXT` listener with no authorizer is a deployment
choice, not a defect.
- **A trusted principal performing an authorized operation.** Admin and
inter-broker actions by a principal that holds the relevant ACLs — or by a
`super.users` entry — are expected behaviour.
- **Findings in development and test tooling.** Issues in `tools`, `bin`,
Trogdor, `tests`, and similar are out of scope (see Development and Test
Tooling above).
-- **Kafka Streams application-level issues.** Streams runs inside the user's
application, so its security boundary is the application's, not the broker's
(see the [Kafka Streams security model](security-model-streams)).
+- **Kafka Streams application-level issues.** Streams runs inside the user's
application, so its security boundary is the application's, not the broker's
(see the [Kafka Streams security model](../security-model-streams)).
## Reporting Security Issues
diff --git a/docs/streams/developer-guide/app-reset-tool.md
b/docs/streams/developer-guide/app-reset-tool.md
index d5d776e462b..3249e2d8366 100644
--- a/docs/streams/developer-guide/app-reset-tool.md
+++ b/docs/streams/developer-guide/app-reset-tool.md
@@ -28,7 +28,7 @@ type: docs
You can reset an application and force it to reprocess its data from scratch
by using the application reset tool. This can be useful for development and
testing, or when fixing bugs.
-The application reset tool handles the Kafka Streams [user
topics](manage-topics.html#streams-developer-guide-topics-user) (input, and
output) and [internal
topics](manage-topics.html#streams-developer-guide-topics-internal) differently
when resetting the application.
+The application reset tool handles the Kafka Streams [user
topics](../manage-topics#streams-developer-guide-topics-user) (input, and
output) and [internal
topics](../manage-topics#streams-developer-guide-topics-internal) differently
when resetting the application.
Here's what the application reset tool does for each topic type:
diff --git a/docs/streams/developer-guide/config-streams.md
b/docs/streams/developer-guide/config-streams.md
index 80c13125a9b..5c006ddaf8a 100644
--- a/docs/streams/developer-guide/config-streams.md
+++ b/docs/streams/developer-guide/config-streams.md
@@ -1200,7 +1200,7 @@ Added to a windows maintainMs to ensure data is not
deleted from the log prematu
<tr>
<td>
-window.size.ms (Deprecated. See [Window Serdes](datatypes.html#window-serdes)
for alternatives.)
+window.size.ms (Deprecated. See [Window Serdes](../datatypes#window-serdes)
for alternatives.)
</td>
<td>
@@ -1217,7 +1217,7 @@ Sets window size for the deserializer in order to
calculate window end times.
<tr>
<td>
-windowed.inner.class.serde (Deprecated. See [Window
Serdes](datatypes.html#window-serdes) for alternatives.)
+windowed.inner.class.serde (Deprecated. See [Window
Serdes](../datatypes#window-serdes) for alternatives.)
</td>
<td>
@@ -1381,7 +1381,7 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
>
>
-> This is discussed in more detail in [Data types and
serialization](datatypes.html#streams-developer-guide-serdes).
+> This is discussed in more detail in [Data types and
serialization](../datatypes#streams-developer-guide-serdes).
### default.value.serde
@@ -1392,7 +1392,7 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
>
>
-> This is discussed in more detail in [Data types and
serialization](datatypes.html#streams-developer-guide-serdes).
+> This is discussed in more detail in [Data types and
serialization](../datatypes#streams-developer-guide-serdes).
### dsl.store.format {#dsl-store-format}
@@ -1493,7 +1493,7 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
### num.standby.replicas
-> The number of standby replicas. Standby replicas are shadow copies of local
state stores. Kafka Streams attempts to create the specified number of replicas
per store and keep them up to date as long as there are enough instances
running. 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 chan [...]
+> The number of standby replicas. Standby replicas are shadow copies of local
state stores. Kafka Streams attempts to create the specified number of replicas
per store and keep them up to date as long as there are enough instances
running. 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 chan [...]
>
> Recommendation:
> Increase the number of standbys to 1 to get instant fail-over, i.e.,
> high-availability. Increasing the number of standbys requires more
> client-side storage space. For example, with 1 standby, 2x space is required.
@@ -1503,7 +1503,7 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
### num.stream.threads
-> This specifies the number of stream threads in an instance of the Kafka
Streams application. The stream processing code runs in these threads. For more
information about Kafka Streams threading model, see [Threading
Model](../architecture.html#streams_architecture_threads).
+> This specifies the number of stream threads in an instance of the Kafka
Streams application. The stream processing code runs in these threads. For more
information about Kafka Streams threading model, see [Threading
Model](../../architecture#streams_architecture_threads).
### probing.rebalance.interval.ms
@@ -1634,7 +1634,7 @@ Serde for the inner class of a windowed record. Must
implement the `Serde` inter
### processing.guarantee
-> The processing guarantee that should be used. Possible values are
`"at_least_once"` (default) and `"exactly_once_v2"` (for EOS version 2).
Deprecated config options are `"exactly_once"` (for EOS alpha), and
`"exactly_once_beta"` (for EOS version 2). Using `"exactly_once_v2"` (or the
deprecated `"exactly_once_beta"`) requires broker version 2.5 or newer, while
using the deprecated `"exactly_once"` requires broker version 0.11.0 or newer.
Note that if exactly-once processing is enabled, [...]
+> The processing guarantee that should be used. Possible values are
`"at_least_once"` (default) and `"exactly_once_v2"` (for EOS version 2).
Deprecated config options are `"exactly_once"` (for EOS alpha), and
`"exactly_once_beta"` (for EOS version 2). Using `"exactly_once_v2"` (or the
deprecated `"exactly_once_beta"`) requires broker version 2.5 or newer, while
using the deprecated `"exactly_once"` requires broker version 0.11.0 or newer.
Note that if exactly-once processing is enabled, [...]
>
> Recommendation:
> While it is technically possible to use EOS with any replication factor,
> using a replication factor lower than 3 effectively voids EOS. Thus it is
> strongly recommended to use a replication factor of 3 (together with
> `min.in.sync.replicas=2`). This recommendation applies to all topics (i.e.
> `__transaction_state`, `__consumer_offsets`, Kafka Streams internal topics,
> and user topics).
diff --git a/docs/streams/developer-guide/datatypes.md
b/docs/streams/developer-guide/datatypes.md
index 4e246cf7f1b..dd25b1c697d 100644
--- a/docs/streams/developer-guide/datatypes.md
+++ b/docs/streams/developer-guide/datatypes.md
@@ -80,7 +80,7 @@ If you want to override serdes selectively, i.e., keep the
defaults for some fie
KStream<String, Long> userCountByRegion = ...;
userCountByRegion.to("RegionCountsTopic",
Produced.valueSerde(Serdes.Long()));
-If some of your incoming records are corrupted or ill-formatted, they will
cause the deserializer class to report an error. Since 1.0.x we have introduced
an `DeserializationExceptionHandler` interface which allows you to customize
how to handle such records. The customized implementation of the interface can
be specified via the `StreamsConfig`. For more details, please feel free to
read the [Configuring a Streams
Application](config-streams.html#default-deserialization-exception-handle [...]
+If some of your incoming records are corrupted or ill-formatted, they will
cause the deserializer class to report an error. Since 1.0.x we have introduced
an `DeserializationExceptionHandler` interface which allows you to customize
how to handle such records. The customized implementation of the interface can
be specified via the `StreamsConfig`. For more details, please feel free to
read the [Configuring a Streams
Application](../config-streams#default-deserialization-exception-handler) [...]
# Available Serdes
@@ -306,9 +306,9 @@ If you need to implement custom Serdes, your best starting
point is to take a lo
-# Kafka Streams DSL for Scala Implicit Serdes[](scala-dsl-serdes "Permalink to
this headline")
+# Kafka Streams DSL for Scala Implicit Serdes[](#scala-dsl-serdes "Permalink
to this headline")
-When using the [Kafka Streams DSL for Scala](dsl-api.html#scala-dsl) you're
not required to configure a default Serdes. In fact, it's not supported. Serdes
are instead provided implicitly by default implementations for common primitive
datatypes. See the [Implicit Serdes](dsl-api.html#scala-dsl-implicit-serdes)
and [User-Defined Serdes](dsl-api.html#scala-dsl-user-defined-serdes) sections
in the DSL API documentation for details
+When using the [Kafka Streams DSL for Scala](../dsl-api#scala-dsl) you're not
required to configure a default Serdes. In fact, it's not supported. Serdes are
instead provided implicitly by default implementations for common primitive
datatypes. See the [Implicit Serdes](../dsl-api#scala-dsl-implicit-serdes) and
[User-Defined Serdes](../dsl-api#scala-dsl-user-defined-serdes) sections in the
DSL API documentation for details
* [Documentation](/documentation)
* [Kafka Streams](/documentation/streams)
diff --git a/docs/streams/developer-guide/dsl-api.md
b/docs/streams/developer-guide/dsl-api.md
index 6f5fc67f555..6ed4e375784 100644
--- a/docs/streams/developer-guide/dsl-api.md
+++ b/docs/streams/developer-guide/dsl-api.md
@@ -58,18 +58,18 @@ The Kafka Streams DSL (Domain Specific Language) is built
on top of the Streams
# Overview
-In comparison to the [Processor
API](processor-api.html#streams-developer-guide-processor-api), only the DSL
supports:
+In comparison to the [Processor
API](../processor-api#streams-developer-guide-processor-api), only the DSL
supports:
- * Built-in abstractions for [streams and
tables](../core-concepts.html#streams_concepts_duality) in the form of KStream,
KTable, and GlobalKTable. Having first-class support for streams and tables is
crucial because, in practice, most use cases require not just either streams or
databases/tables, but a combination of both. For example, if your use case is
to create a customer 360-degree view that is updated in real-time, what your
application will be doing is transforming many input _s [...]
+ * Built-in abstractions for [streams and
tables](../../core-concepts#streams_concepts_duality) in the form of KStream,
KTable, and GlobalKTable. Having first-class support for streams and tables is
crucial because, in practice, most use cases require not just either streams or
databases/tables, but a combination of both. For example, if your use case is
to create a customer 360-degree view that is updated in real-time, what your
application will be doing is transforming many input _str [...]
* Declarative, functional programming style with stateless transformations
(e.g. `map` and `filter`) as well as stateful transformations such as
aggregations (e.g. `count` and `reduce`), joins (e.g. `leftJoin`), and
windowing (e.g. session windows).
-With the DSL, you can define [processor
topologies](../core-concepts.html#streams_topology) (i.e., the logical
processing plan) in your application. The steps to accomplish this are:
+With the DSL, you can define [processor
topologies](../../core-concepts#streams_topology) (i.e., the logical processing
plan) in your application. The steps to accomplish this are:
1. Specify one or more input streams that are read from Kafka topics.
2. Compose transformations on these streams.
- 3. Write the resulting output streams back to Kafka topics, or expose the
processing results of your application directly to other applications through
[interactive
queries](interactive-queries.html#streams-developer-guide-interactive-queries)
(e.g., via a REST API).
+ 3. Write the resulting output streams back to Kafka topics, or expose the
processing results of your application directly to other applications through
[interactive
queries](../interactive-queries#streams-developer-guide-interactive-queries)
(e.g., via a REST API).
@@ -105,7 +105,7 @@ If your stream processing application were to sum the
values per user, it would
However, it would not be safe to enable log compaction in the case of a
KStream because, as soon as log compaction would begin purging older data
records of the same key, it would break the semantics of the data. To pick up
the illustration example again, you'd suddenly get a `3` for `alice` instead of
a `4` because log compaction would have removed the `("alice", 1)` data record.
Hence log compaction is perfectly safe for a KTable (changelog stream) but it
is a mistake for a KStream (re [...]
-We have already seen an example of a changelog stream in the section [streams
and tables](../core-concepts.html#streams_concepts_duality). Another example
are change data capture (CDC) records in the changelog of a relational
database, representing which row in a database table was inserted, updated, or
deleted.
+We have already seen an example of a changelog stream in the section [streams
and tables](../../core-concepts#streams_concepts_duality). Another example are
change data capture (CDC) records in the changelog of a relational database,
representing which row in a database table was inserted, updated, or deleted.
KTable also provides an ability to look up _current_ values of data records by
keys. This table-lookup functionality is available through **join operations**
(see also **Joining** in the Developer Guide) as well as through **Interactive
Queries**.
@@ -185,9 +185,9 @@ In the case of a KStream, the local KStream instance of
every application instan
Serdes.Long() /* value serde */
);
-If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.html#streams-developer-guide-configuration) are
used.
+If you do not specify Serdes explicitly, the default Serdes from the
[configuration](../config-streams#streams-developer-guide-configuration) are
used.
-You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](datatypes.html#streams-developer-guide-serdes).
+You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](../datatypes#streams-developer-guide-serdes).
Several variants of `stream` exist. For example, you can specify a regex
pattern for input topics to read from (note that all matching topics will be
part of the same input topic group, and the work will not be parallelized for
different topics if subscribed to in this way).
@@ -212,11 +212,11 @@ Reads the specified Kafka input topic into a KTable. The
topic is interpreted as
In the case of a KTable, the local KTable instance of every application
instance will be populated with data from only **a subset** of the partitions
of the input topic. Collectively, across all application instances, all input
topic partitions are read and processed.
-You must provide a name for the table (more precisely, for the internal [state
store](../architecture.html#streams_architecture_state) that backs the table).
This is required for supporting [interactive
queries](interactive-queries.html#streams-developer-guide-interactive-queries)
against the table. When a name is not provided the table will not be queryable
and an internal name will be provided for the state store.
+You must provide a name for the table (more precisely, for the internal [state
store](../../architecture#streams_architecture_state) that backs the table).
This is required for supporting [interactive
queries](../interactive-queries#streams-developer-guide-interactive-queries)
against the table. When a name is not provided the table will not be queryable
and an internal name will be provided for the state store.
-If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.html#streams-developer-guide-configuration) are
used.
+If you do not specify Serdes explicitly, the default Serdes from the
[configuration](../config-streams#streams-developer-guide-configuration) are
used.
-You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](datatypes.html#streams-developer-guide-serdes).
+You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](../datatypes#streams-developer-guide-serdes).
Several variants of `table` exist, for example to specify the
`auto.offset.reset` policy to be used when reading from the input topic.
@@ -241,7 +241,7 @@ Reads the specified Kafka input topic into a GlobalKTable.
The topic is interpre
In the case of a GlobalKTable, the local GlobalKTable instance of every
application instance will be populated with data from **all** the partitions of
the input topic.
-You must provide a name for the table (more precisely, for the internal [state
store](../architecture.html#streams_architecture_state) that backs the table).
This is required for supporting [interactive
queries](interactive-queries.html#streams-developer-guide-interactive-queries)
against the table. When a name is not provided the table will not be queryable
and an internal name will be provided for the state store.
+You must provide a name for the table (more precisely, for the internal [state
store](../../architecture#streams_architecture_state) that backs the table).
This is required for supporting [interactive
queries](../interactive-queries#streams-developer-guide-interactive-queries)
against the table. When a name is not provided the table will not be queryable
and an internal name will be provided for the state store.
import org.apache.kafka.common.serialization.Serdes;
@@ -258,7 +258,7 @@ You must provide a name for the table (more precisely, for
the internal [state s
.withValueSerde(Serdes.Long()) /* value serde */
);
-You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](datatypes.html#streams-developer-guide-serdes).
+You **must specify Serdes explicitly** if the key or value types of the
records in the Kafka input topics do not match the configured default Serdes.
For information about configuring default Serdes, available Serdes, and
implementing your own custom Serdes see [Data Types and
Serialization](../datatypes#streams-developer-guide-serdes).
Several variants of `globalTable` exist to e.g. specify explicit Serdes.
@@ -271,7 +271,7 @@ The KStream and KTable interfaces support a variety of
transformation operations
Some KStream transformations may generate one or more KStream objects, for
example: \- `filter` and `map` on a KStream will generate another KStream \-
`split` on KStream can generate multiple KStreams
-Some others may generate a KTable object, for example an aggregation of a
KStream also yields a KTable. This allows Kafka Streams to continuously update
the computed value upon arrivals of [out-of-order
records](../core-concepts.html#streams_concepts_aggregations) after it has
already been produced to the downstream transformation operators.
+Some others may generate a KTable object, for example an aggregation of a
KStream also yields a KTable. This allows Kafka Streams to continuously update
the computed value upon arrivals of [out-of-order
records](../../core-concepts#streams_concepts_aggregations) after it has
already been produced to the downstream transformation operators.
All KTable transformation operations can only generate another KTable.
However, the Kafka Streams DSL does provide a special function that converts a
KTable representation into a KStream. All of these transformation methods can
be chained together to compose a complex processor topology.
@@ -284,7 +284,7 @@ These transformation operations are described in the
following subsections:
## Stateless transformations
-Stateless transformations do not require state for processing and they do not
require a state store associated with the stream processor. Kafka 0.11.0 and
later allows you to materialize the result from a stateless `KTable`
transformation. This allows the result to be queried through [interactive
queries](interactive-queries.html#streams-developer-guide-interactive-queries).
To materialize a `KTable`, each of the below stateless operations [can be
augmented](interactive-queries.html#stre [...]
+Stateless transformations do not require state for processing and they do not
require a state store associated with the stream processor. Kafka 0.11.0 and
later allows you to materialize the result from a stateless `KTable`
transformation. This allows the result to be queried through [interactive
queries](../interactive-queries#streams-developer-guide-interactive-queries).
To materialize a `KTable`, each of the below stateless operations [can be
augmented](../interactive-queries#streams- [...]
<table>
<tr>
@@ -915,7 +915,7 @@ Kafka Streams will manage the topic for `repartition()`.
Generated topic is trea
## Stateful transformations
-Stateful transformations depend on state for processing inputs and producing
outputs and require a [state
store](../architecture.html#streams_architecture_state) associated with the
stream processor. For example, in aggregating operations, a windowing state
store is used to collect the latest aggregation results per window. In join
operations, a windowing state store is used to collect all of the records
received so far within the defined window boundary.
+Stateful transformations depend on state for processing inputs and producing
outputs and require a [state
store](../../architecture#streams_architecture_state) associated with the
stream processor. For example, in aggregating operations, a windowing state
store is used to collect the latest aggregation results per window. In join
operations, a windowing state store is used to collect all of the records
received so far within the defined window boundary.
**Note:** Following store types are used regardless of the possibly specified
type (via the parameter `materialized`):
@@ -925,7 +925,7 @@ Stateful transformations depend on state for processing
inputs and producing out
-**Headers-aware state stores
([KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G)):** Set
[`dsl.store.format=HEADERS`](config-streams.html#dsl-store-format) to make
supported DSL operators use headers-aware state stores. These stores can keep
record headers together with the value and timestamp.
+**Headers-aware state stores
([KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G)):** Set
[`dsl.store.format=HEADERS`](../config-streams#dsl-store-format) to make
supported DSL operators use headers-aware state stores. These stores can keep
record headers together with the value and timestamp.
This config only changes the state store format. It does not define how DSL
operators create headers for output records. Current behavior is:
@@ -935,7 +935,7 @@ This config only changes the state store format. It does
not define how DSL oper
A follow-up KIP will define how DSL result headers are computed.
-Note, that state stores are fault-tolerant. In case of failure, Kafka Streams
guarantees to fully restore all state stores prior to resuming the processing.
See [Fault Tolerance](../architecture.html#streams_architecture_recovery) for
further information.
+Note, that state stores are fault-tolerant. In case of failure, Kafka Streams
guarantees to fully restore all state stores prior to resuming the processing.
See [Fault Tolerance](../../architecture#streams_architecture_recovery) for
further information.
Available stateful transformations in the DSL include:
@@ -1348,7 +1348,7 @@ See the example at the bottom of this section for a
visualization of the aggrega
**Note**
-**Impact of record caches** : For illustration purposes, the column "KTable
`aggregated`" below shows the table's state changes over time in a very
granular way. In practice, you would observe state changes in such a granular
way only when [record
caches](memory-mgmt.html#streams-developer-guide-memory-management-record-cache)
are disabled (default: enabled). When record caches are enabled, what might
happen for example is that the output results of the rows with timestamps 4 and
5 would [...]
+**Impact of record caches** : For illustration purposes, the column "KTable
`aggregated`" below shows the table's state changes over time in a very
granular way. In practice, you would observe state changes in such a granular
way only when [record
caches](../memory-mgmt#streams-developer-guide-memory-management-record-cache)
are disabled (default: enabled). When record caches are enabled, what might
happen for example is that the output results of the rows with timestamps 4 and
5 would b [...]
<table>
<tr>
@@ -1608,7 +1608,7 @@ State
**Note**
-**Impact of record caches** : For illustration purposes, the column "KTable
`aggregated`" below shows the table's state changes over time in a very
granular way. In practice, you would observe state changes in such a granular
way only when [record
caches](memory-mgmt.html#streams-developer-guide-memory-management-record-cache)
are disabled (default: enabled). When record caches are enabled, what might
happen for example is that the output results of the rows with timestamps 4 and
5 would [...]
+**Impact of record caches** : For illustration purposes, the column "KTable
`aggregated`" below shows the table's state changes over time in a very
granular way. In practice, you would observe state changes in such a granular
way only when [record
caches](../memory-mgmt#streams-developer-guide-memory-management-record-cache)
are disabled (default: enabled). When record caches are enabled, what might
happen for example is that the output results of the rows with timestamps 4 and
5 would b [...]
<table>
<tr>
@@ -2127,7 +2127,7 @@ There are two exceptions where co-partitioning is not
required. For KStream-Glob
KStream-KStream joins are always windowed joins, because otherwise the size of
the internal state store used to perform the join - e.g., a sliding window or
"buffer" - would grow indefinitely.
-**Note on headers-aware state stores:** With
[`dsl.store.format=HEADERS`](config-streams.html#dsl-store-format), inner
stream-stream joins use headers-aware join window stores. Left and outer
stream-stream joins also use a separate buffer store for not-yet-matched
records, and that buffer is not headers-aware. Records that pass through this
buffer lose their headers.
+**Note on headers-aware state stores:** With
[`dsl.store.format=HEADERS`](../config-streams#dsl-store-format), inner
stream-stream joins use headers-aware join window stores. Left and outer
stream-stream joins also use a separate buffer store for not-yet-matched
records, and that buffer is not headers-aware. Records that pass through this
buffer lose their headers.
Join output records do not get computed or merged headers. The current
forwarding path may carry the headers from the record that triggered the
output. For stream-stream joins it's important to highlight that a new input
record on one side will produce a join output _for each_ matching record on the
other side, and there can be _multiple_ such matching records in a given join
window (cf. the row with timestamp 15 in the join semantics table below, for
example).
@@ -4462,7 +4462,7 @@ Windowing lets you control how to group records that have
the same key for state
A related operation is grouping, which groups all records that have the same
key to ensure that data is properly partitioned ("keyed") for subsequent
operations. Once grouped, windowing allows you to further sub-group the records
of a key.
-For example, in join operations, a windowing state store is used to store all
the records received so far within the defined window boundary. In aggregating
operations, a windowing state store is used to store the latest aggregation
results per window. Old records in the state store are purged after the
specified [window retention
period](../core-concepts.html#streams_concepts_windowing). Kafka Streams
guarantees to keep a window for at least this specified time; the default value
is one [...]
+For example, in join operations, a windowing state store is used to store all
the records received so far within the defined window boundary. In aggregating
operations, a windowing state store is used to store the latest aggregation
results per window. Old records in the state store are purged after the
specified [window retention
period](../../core-concepts#streams_concepts_windowing). Kafka Streams
guarantees to keep a window for at least this specified time; the default value
is one d [...]
The DSL supports the following types of windows:
@@ -4673,7 +4673,7 @@ The key parts of this program are:
One thing to note is that suppression is just like any other Kafka Streams
operator, so you can build a topology with two branches emerging from the
`count`, one suppressed, and one not, or even multiple differently configured
suppressions. This allows you to apply suppressions where they are needed and
otherwise rely on the default continuous update behavior.
-**Note on headers-aware state stores:** `suppress()` uses an in-memory buffer
that is not headers-aware. Record headers attached to upstream records are not
preserved across the suppression boundary, even when
[`dsl.store.format=HEADERS`](config-streams.html#dsl-store-format) is set
globally per [KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G).
+**Note on headers-aware state stores:** `suppress()` uses an in-memory buffer
that is not headers-aware. Record headers attached to upstream records are not
preserved across the suppression boundary, even when
[`dsl.store.format=HEADERS`](../config-streams#dsl-store-format) is set
globally per [KIP-1285](https://cwiki.apache.org/confluence/x/4ow8G).
For more detailed information, see the JavaDoc on the `Suppressed` config
object and [KIP-328](https://cwiki.apache.org/confluence/x/sQU0BQ "KIP-328").
@@ -5621,7 +5621,7 @@ By automating the detection of popular pages, the system
eliminates the need for
}
}
-Naming Operators in a Streams DSL application Kafka Streams allows you to
[name processors](dsl-topology-naming.html) created via the Streams DSL
+Naming Operators in a Streams DSL application Kafka Streams allows you to
[name processors](../dsl-topology-naming) created via the Streams DSL
# Controlling KTable emit rate
@@ -5629,7 +5629,7 @@ A KTable is logically a continuously updated table. These
updates make their way
However, some applications need to take other actions, such as calling out to
external systems, and therefore need to exercise some control over the rate of
invocations, for example of `KStream#foreach`.
-Rather than achieving this as a side-effect of the [KTable record
cache](memory-mgmt.html#streams-developer-guide-memory-management-record-cache),
you can directly impose a rate limit via the `KTable#suppress` operator.
+Rather than achieving this as a side-effect of the [KTable record
cache](../memory-mgmt#streams-developer-guide-memory-management-record-cache),
you can directly impose a rate limit via the `KTable#suppress` operator.
For example:
@@ -5713,9 +5713,9 @@ Description
When to provide serdes explicitly:
- * If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.html#streams-developer-guide-configuration) are
used.
+ * If you do not specify Serdes explicitly, the default Serdes from the
[configuration](../config-streams#streams-developer-guide-configuration) are
used.
* You **must specify Serdes explicitly** via the `Produced` class if the key
and/or value types of the `KStream` do not match the configured default Serdes.
- * See [Data Types and
Serialization](datatypes.html#streams-developer-guide-serdes) for information
about configuring default Serdes, available Serdes, and implementing your own
custom Serdes.
+ * See [Data Types and
Serialization](../datatypes#streams-developer-guide-serdes) for information
about configuring default Serdes, available Serdes, and implementing your own
custom Serdes.
@@ -5746,11 +5746,11 @@ Another variant of `to` exists that enables you to
dynamically choose which topi
**Note**
-**When you want to write to systems other than Kafka:** Besides writing the
data back to Kafka, you can also apply a custom processor as a stream sink at
the end of the processing to, for example, write to external databases. First,
doing so is not a recommended pattern - we strongly suggest to use the [Kafka
Connect API](../../connect/index.html#kafka-connect) instead. However, if you
do use such a sink processor, please be aware that it is now your
responsibility to guarantee message d [...]
+**When you want to write to systems other than Kafka:** Besides writing the
data back to Kafka, you can also apply a custom processor as a stream sink at
the end of the processing to, for example, write to external databases. First,
doing so is not a recommended pattern - we strongly suggest to use the [Kafka
Connect API](../../../kafka-connect/#kafka-connect) instead. However, if you do
use such a sink processor, please be aware that it is now your responsibility
to guarantee message de [...]
# Testing a Streams application
-Kafka Streams comes with a `test-utils` module to help you test your
application [here](testing.html).
+Kafka Streams comes with a `test-utils` module to help you test your
application [here](../testing).
# Kafka Streams DSL for Scala
diff --git a/docs/streams/developer-guide/interactive-queries.md
b/docs/streams/developer-guide/interactive-queries.md
index d3fed2f4a14..53d3100cf40 100644
--- a/docs/streams/developer-guide/interactive-queries.md
+++ b/docs/streams/developer-guide/interactive-queries.md
@@ -31,7 +31,7 @@ Interactive queries allow you to leverage the state of your
application from out
-The full state of your application is typically [split across many distributed
instances of your
application](../architecture.html#streams_architecture_state), and across many
state stores that are managed locally by these application instances.
+The full state of your application is typically [split across many distributed
instances of your application](../../architecture#streams_architecture_state),
and across many state stores that are managed locally by these application
instances.
@@ -125,7 +125,7 @@ Not supported (you must configure)
# Querying local state stores for an app instance
{#querying-local-state-stores-for-an-app-instance}
-A Kafka Streams application typically runs on multiple instances. The state
that is locally available on any given instance is only a subset of the
[application's entire state](../architecture.html#streams-architecture-state).
Querying the local stores on an instance will only return data locally
available on that particular instance.
+A Kafka Streams application typically runs on multiple instances. The state
that is locally available on any given instance is only a subset of the
[application's entire state](../../architecture#streams-architecture-state).
Querying the local stores on an instance will only return data locally
available on that particular instance.
The method `KafkaStreams#store(...)` finds an application instance's local
state stores by name and type. Note that interactive queries are not supported
for [versioned state
stores](/{version}/streams/developer-guide/processor-api/#versioned-key-value-state-stores)
at this time.
@@ -396,7 +396,7 @@ There are many ways to add an RPC layer. The only
requirements are that the RPC
## Exposing the RPC endpoints of your application
-To enable remote state store discovery in a distributed Kafka Streams
application, you must set the [configuration
property](config-streams.html#streams-developer-guide-required-configs) in the
config properties. The `application.server` property defines a unique
`host:port` pair that points to the RPC endpoint of the respective instance of
a Kafka Streams application. The value of this configuration property will vary
across the instances of your application. When this property is set, [...]
+To enable remote state store discovery in a distributed Kafka Streams
application, you must set the [configuration
property](../config-streams#streams-developer-guide-required-configs) in the
config properties. The `application.server` property defines a unique
`host:port` pair that points to the RPC endpoint of the respective instance of
a Kafka Streams application. The value of this configuration property will vary
across the instances of your application. When this property is set, Ka [...]
**Tip**
diff --git a/docs/streams/developer-guide/manage-topics.md
b/docs/streams/developer-guide/manage-topics.md
index 1a16e7ab229..aac4ab745e4 100644
--- a/docs/streams/developer-guide/manage-topics.md
+++ b/docs/streams/developer-guide/manage-topics.md
@@ -39,7 +39,7 @@ Input topics
Output topics
Topics that are specified via sink processors in the application's
topology; e.g. via `KStream#to()`, `KTable.to()` and `Topology#addSink()`.
-User topics must be created and manually managed ahead of time (e.g., via the
[topic tools](../../kafka/post-deployment.html#kafka-operations-admin)). If
user topics are shared among multiple applications for reading and writing, the
application users must coordinate topic management. If user topics are
centrally managed, then application users would not need to manage topics
themselves but simply obtain access to them.
+User topics must be created and manually managed ahead of time (e.g., via the
[topic
tools](../../../operations/basic-kafka-operations/#kafka-operations-admin)). If
user topics are shared among multiple applications for reading and writing, the
application users must coordinate topic management. If user topics are
centrally managed, then application users would not need to manage topics
themselves but simply obtain access to them.
Note
@@ -54,7 +54,7 @@ You should not use the auto-create topic feature on the
brokers to create user t
Internal topics are used internally by the Kafka Streams application while
executing, for example the changelog topics for state stores. These topics are
created by the application and are only used by that stream application.
-If security is enabled on the Kafka brokers, you must grant the underlying
clients admin permissions so that they can create internal topics set. For more
information, see [Streams
Security](security.html#streams-developer-guide-security).
+If security is enabled on the Kafka brokers, you must grant the underlying
clients admin permissions so that they can create internal topics set. For more
information, see [Streams
Security](../security#streams-developer-guide-security).
Note
diff --git a/docs/streams/developer-guide/memory-mgmt.md
b/docs/streams/developer-guide/memory-mgmt.md
index c63f997d16e..b79a7348ddd 100644
--- a/docs/streams/developer-guide/memory-mgmt.md
+++ b/docs/streams/developer-guide/memory-mgmt.md
@@ -38,20 +38,20 @@ The record caches are implemented slightly different in the
DSL and Processor AP
You can specify the total memory (RAM) size of the record cache for an
instance of the processing topology. It is leveraged by the following `KTable`
instances:
* Source `KTable`: `KTable` instances that are created via
`StreamsBuilder#table()` or `StreamsBuilder#globalTable()`.
- * Aggregation `KTable`: instances of `KTable` that are created as a result
of [aggregations](dsl-api.html#streams-developer-guide-dsl-aggregating).
+ * Aggregation `KTable`: instances of `KTable` that are created as a result
of [aggregations](../dsl-api#streams-developer-guide-dsl-aggregating).
For such `KTable` instances, the record cache is used for:
- * Internal caching and compacting of output records before they are written
by the underlying stateful [processor
node](../core-concepts#streams_processor_node) to its internal state stores.
- * Internal caching and compacting of output records before they are
forwarded from the underlying stateful [processor
node](../core-concepts#streams_processor_node) to any of its downstream
processor nodes.
+ * Internal caching and compacting of output records before they are written
by the underlying stateful [processor
node](../../core-concepts#streams_processor_node) to its internal state stores.
+ * Internal caching and compacting of output records before they are
forwarded from the underlying stateful [processor
node](../../core-concepts#streams_processor_node) to any of its downstream
processor nodes.
Use the following example to understand the behaviors with and without record
caching. In this example, the input is a `KStream<String, Integer>` with the
records `<K,V>: <A, 1>, <D, 5>, <A, 20>, <A, 300>`. The focus in this example
is on the records with key == `A`.
- * An [aggregation](dsl-api.html#streams-developer-guide-dsl-aggregating)
computes the sum of record values, grouped by key, for the input and returns a
`KTable<String, Integer>`.
+ * An [aggregation](../dsl-api#streams-developer-guide-dsl-aggregating)
computes the sum of record values, grouped by key, for the input and returns a
`KTable<String, Integer>`.
> * **Without caching** : a sequence of output records is emitted for key
> `A` that represent changes in the resulting aggregation table. The
> parentheses (`()`) denote changes, the left number is the new aggregate
> value and the right number is the old aggregate value: `<A, (1, null)>, <A,
> (21, 1)>, <A, (321, 21)>`.
> * **With caching** : a single output record is emitted for key `A` that
> would likely be compacted in the cache, leading to a single output record of
> `<A, (321, null)>`. This record is written to the aggregation's internal
> state store and forwarded to any downstream operations.
@@ -112,7 +112,7 @@ You can specify the total memory (RAM) size of the record
cache for an instance
The record cache in the Processor API does not cache or compact any output
records that are being forwarded downstream. This means that all downstream
processor nodes can see all records, whereas the state stores see a reduced
number of records. This does not impact correctness of the system, but is a
performance optimization for the state stores. For example, with the Processor
API you can store a record in a state store while forwarding a different value
downstream.
-Following from the example first shown in section [State
Stores](processor-api.html#streams-developer-guide-state-store), to disable
caching, you can add the `withCachingDisabled` call (note that caches are
enabled by default, however there is an explicit `withCachingEnabled` call).
+Following from the example first shown in section [State
Stores](../processor-api#streams-developer-guide-state-store), to disable
caching, you can add the `withCachingDisabled` call (note that caches are
enabled by default, however there is an explicit `withCachingEnabled` call).
StoreBuilder countStoreBuilder =
@@ -122,7 +122,7 @@ Following from the example first shown in section [State
Stores](processor-api.h
Serdes.Long())
.withCachingEnabled();
-Record caches are not supported for [versioned state
stores](processor-api.html#streams-developer-guide-state-store-versioned).
+Record caches are not supported for [versioned state
stores](../processor-api#streams-developer-guide-state-store-versioned).
To avoid reading stale data, you can `flush()` the store before creating the
iterator. Note, that flushing too often can lead to performance degradation if
RocksDB is used, so we advise to avoid flushing manually in general.
diff --git a/docs/streams/developer-guide/processor-api.md
b/docs/streams/developer-guide/processor-api.md
index 5bfb25cec4a..a35e4f6e43d 100644
--- a/docs/streams/developer-guide/processor-api.md
+++ b/docs/streams/developer-guide/processor-api.md
@@ -37,13 +37,13 @@ The Processor API can be used to implement both
**stateless** as well as **state
**Tip**
-**Combining the DSL and the Processor API:** You can combine the convenience
of the DSL with the power and flexibility of the Processor API as described in
the section [Applying processors (Processor API
integration)](dsl-api.html#streams-developer-guide-dsl-process).
+**Combining the DSL and the Processor API:** You can combine the convenience
of the DSL with the power and flexibility of the Processor API as described in
the section [Applying processors (Processor API
integration)](../dsl-api#streams-developer-guide-dsl-process).
For a complete list of available API functionality, see the
[Streams](/{version}/javadoc/org/apache/kafka/streams/package-summary.html) API
docs.
# Defining a Stream Processor
-A [stream processor](../core-concepts.html#streams_processor_node) is a node
in the processor topology that represents a single processing step. With the
Processor API, you can define arbitrary stream processors that processes one
received record at a time, and connect these processors with their associated
state stores to compose the processor topology.
+A [stream processor](../../core-concepts#streams_processor_node) is a node in
the processor topology that represents a single processing step. With the
Processor API, you can define arbitrary stream processors that processes one
received record at a time, and connect these processors with their associated
state stores to compose the processor topology.
You can define a customized stream processor by implementing the `Processor`
interface, which provides the `process()` API method. The `process()` method is
called on each of the received records.
@@ -53,7 +53,7 @@ The `Processor` interface takes four generic parameters:
`KIn, VIn, KOut, VOut`.
Both the `Processor#process()` and the `ProcessorContext#forward()` methods
handle records in the form of the `Record<K, V>` data class. This class gives
you access to the main components of a Kafka record: the key, value, timestamp
and headers. When forwarding records, you can use the constructor to create a
new `Record` from scratch, or you can use the convenience builder methods to
replace one of the `Record`'s properties and copy over the rest. For example,
`inputRecord.withValue(new [...]
-In addition to handling incoming records via `Processor#process()`, you have
the option to schedule periodic invocation (called "punctuation") in your
processor's `init()` method by calling `ProcessorContext#schedule()` and
passing it a `Punctuator`. The `PunctuationType` determines what notion of time
is used for the punctuation scheduling: either
[stream-time](../core-concepts.html#streams_time) or wall-clock-time (by
default, stream-time is configured to represent event-time via `Time [...]
+In addition to handling incoming records via `Processor#process()`, you have
the option to schedule periodic invocation (called "punctuation") in your
processor's `init()` method by calling `ProcessorContext#schedule()` and
passing it a `Punctuator`. The `PunctuationType` determines what notion of time
is used for the punctuation scheduling: either
[stream-time](../../core-concepts#streams_time) or wall-clock-time (by default,
stream-time is configured to represent event-time via `Timest [...]
For example, if you schedule a `Punctuator` function every 10 seconds based on
`PunctuationType.STREAM_TIME` and if you process a stream of 60 records with
consecutive timestamps from 1 (first record) to 60 seconds (last record), then
`punctuate()` would be called 6 times. This happens regardless of the time
required to actually process those records. `punctuate()` would be called 6
times regardless of whether processing these 60 records takes a second, a
minute, or an hour.
@@ -118,11 +118,11 @@ The following example `Processor` defines a simple
word-count algorithm and the
# Unit Testing Processors
-Kafka Streams comes with a `test-utils` module to help you write unit tests
for your processors [here](testing.html#unit-testing-processors).
+Kafka Streams comes with a `test-utils` module to help you write unit tests
for your processors [here](../testing#unit-testing-processors).
# State Stores
-To implement a **stateful** `Processor`, you must provide one or more state
stores to the processor (_stateless_ processors do not need state stores).
State stores can be used to remember recently received input records, to track
rolling aggregates, to de-duplicate input records, and more. Another feature of
state stores is that they can be [interactively
queried](interactive-queries.html#streams-developer-guide-interactive-queries)
from other applications, such as a NodeJS-based dashboa [...]
+To implement a **stateful** `Processor`, you must provide one or more state
stores to the processor (_stateless_ processors do not need state stores).
State stores can be used to remember recently received input records, to track
rolling aggregates, to de-duplicate input records, and more. Another feature of
state stores is that they can be [interactively
queried](../interactive-queries#streams-developer-guide-interactive-queries)
from other applications, such as a NodeJS-based dashboard [...]
The available state store types in Kafka Streams have fault tolerance enabled
by default.
@@ -172,7 +172,7 @@ Yes (enabled by default)
* **The recommended store type for most use cases.**
* Stores its data on local disk.
* Storage capacity: managed local state can be larger than the memory (heap
space) of an application instance, but must fit into the available local disk
space.
- * RocksDB settings can be fine-tuned, see [RocksDB
configuration](config-streams.html#streams-developer-guide-rocksdb-config).
+ * RocksDB settings can be fine-tuned, see [RocksDB
configuration](../config-streams#streams-developer-guide-rocksdb-config).
* Available [persistent store
variants](<https://kafka.apache.org/{version}/javadoc/org/apache/kafka/streams/state/Stores.html#persistentKeyValueStore(java.lang.String)>):
plain key-value store (values only — no embedded record timestamp in state),
timestamped key-value store, versioned key-value store, windowed store, session
store. Header-aware variants are also described below.
* Use
[persistentKeyValueStore](<https://kafka.apache.org/{version}/javadoc/org/apache/kafka/streams/state/Stores.html#persistentKeyValueStore(java.lang.String)>)
when you need a persistent plain key-value store (no embedded record
timestamp).
* Use
[persistentTimestampedKeyValueStore](<https://kafka.apache.org/{version}/javadoc/org/apache/kafka/streams/state/Stores.html#persistentTimestampedKeyValueStore(java.lang.String)>)
when you need a persistent key-(value/timestamp) store that supports
put/get/delete and range queries.
@@ -248,7 +248,7 @@ Yes (enabled by default)
## Fault-tolerant State Stores
-To make state stores fault-tolerant and to allow for state store migration
without data loss, a state store can be continuously backed up to a Kafka topic
behind the scenes. For example, to migrate a stateful stream task from one
machine to another when [elastically adding or removing capacity from your
application](running-app.html#streams-developer-guide-execution-scaling). This
topic is sometimes referred to as the state store's associated _changelog
topic_ , or its _changelog_. For e [...]
+To make state stores fault-tolerant and to allow for state store migration
without data loss, a state store can be continuously backed up to a Kafka topic
behind the scenes. For example, to migrate a stateful stream task from one
machine to another when [elastically adding or removing capacity from your
application](../running-app#streams-developer-guide-execution-scaling). This
topic is sometimes referred to as the state store's associated _changelog
topic_ , or its _changelog_. For exa [...]
Fault-tolerant state stores are backed by a
[compacted](https://kafka.apache.org/documentation.html#compaction) changelog
topic. The purpose of compacting this topic is to prevent the topic from
growing indefinitely, to reduce the storage consumed in the associated Kafka
cluster, and to minimize recovery time if a state store needs to be restored
from its changelog topic.
@@ -274,7 +274,7 @@ Example for disabling fault-tolerance:
Attention
-If the changelog is disabled then the attached state store is no longer fault
tolerant and it can't have any [standby
replicas](config-streams.html#streams-developer-guide-standby-replicas).
+If the changelog is disabled then the attached state store is no longer fault
tolerant and it can't have any [standby
replicas](../config-streams#streams-developer-guide-standby-replicas).
Here is an example for enabling fault tolerance, with additional
changelog-topic configuration: You can add any log config from
[kafka.log.LogConfig](https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/log/LogConfig.scala).
Unrecognized configs will be ignored.
@@ -326,7 +326,7 @@ Each versioned store has an associated, fixed-duration
_history retention_ param
History retention also doubles as its _grace period_ , which determines how
far back in time out-of-order writes to the store will be accepted. A versioned
store will not accept writes (inserts, updates, or deletions) if the timestamp
associated with the write is older than the current observed stream time by
more than the grace period. Stream time in this context is tracked
per-partition, rather than per-key, which means it's important that grace
period (i.e., history retention) be set [...]
-Because the memory footprint of versioned key-value stores is higher than that
of non-versioned key-value stores, you may want to adjust your [RocksDB memory
settings](memory-mgmt.html#streams-developer-guide-memory-management-rocksdb)
accordingly. Benchmarking your application with versioned stores is also
advised as performance is expected to be worse than when using non-versioned
stores.
+Because the memory footprint of versioned key-value stores is higher than that
of non-versioned key-value stores, you may want to adjust your [RocksDB memory
settings](../memory-mgmt#streams-developer-guide-memory-management-rocksdb)
accordingly. Benchmarking your application with versioned stores is also
advised as performance is expected to be worse than when using non-versioned
stores.
Versioned stores do not support caching or interactive queries at this time.
Also, window stores and global tables may not be versioned.
@@ -336,7 +336,7 @@ Upgrades are supported from persistent, non-versioned
key-value stores to persis
If you wish to upgrade an application using persistent, non-versioned
key-value stores to use persistent, versioned key-value stores instead, you can
perform the following procedure:
- * Stop all application instances, and [clear any local state
directories](app-reset-tool.html#streams-developer-guide-reset-local-environment)
for the store(s) being upgraded.
+ * Stop all application instances, and [clear any local state
directories](../app-reset-tool#streams-developer-guide-reset-local-environment)
for the store(s) being upgraded.
* Update your application code to use versioned stores where desired.
* Update your changelog topic configs, for the relevant state stores, to set
the value of `min.compaction.lag.ms` to be at least your desired history
retention. History retention plus one day is recommended as buffer for the use
of broker wall clock time during compaction.
* Restart your application instances and allow time for the versioned stores
to rebuild state from changelog.
@@ -424,7 +424,7 @@ In these topologies, the `"Process"` stream processor node
is considered a downs
Note that the `Topology#addProcessor` function takes a `ProcessorSupplier` as
argument, and that the supplier pattern requires that a new `Processor`
instance is returned each time `ProcessorSupplier#get()` is called. Creating a
single `Processor` object and returning the same object reference in
`ProcessorSupplier#get()` would be a violation of the supplier pattern and
leads to runtime exceptions. So remember not to provide a singleton `Processor`
instance to `Topology`. The `ProcessorS [...]
-Now that you have fully defined your processor topology in your application,
you can proceed to [running the Kafka Streams
application](running-app.html#streams-developer-guide-execution).
+Now that you have fully defined your processor topology in your application,
you can proceed to [running the Kafka Streams
application](../running-app#streams-developer-guide-execution).
* [Documentation](/documentation)
* [Kafka Streams](/documentation/streams)
diff --git a/docs/streams/developer-guide/running-app.md
b/docs/streams/developer-guide/running-app.md
index 32310df11ad..edc56543dc9 100644
--- a/docs/streams/developer-guide/running-app.md
+++ b/docs/streams/developer-guide/running-app.md
@@ -40,7 +40,7 @@ You can package your Java application as a fat JAR file and
then start the appli
# from the fat JAR named `path-to-app-fatjar.jar`.
$ java -cp path-to-app-fatjar.jar com.example.MyStreamsApp
-When you start your application you are launching a Kafka Streams instance of
your application. You can run multiple instances of your application. A common
scenario is that there are multiple instances of your application running in
parallel. For more information, see [Parallelism
Model](../architecture.html#streams_architecture_tasks).
+When you start your application you are launching a Kafka Streams instance of
your application. You can run multiple instances of your application. A common
scenario is that there are multiple instances of your application running in
parallel. For more information, see [Parallelism
Model](../../architecture#streams_architecture_tasks).
When the application instance starts running, the defined processor topology
will be initialized as one or more stream tasks. If the processor topology
defines any state stores, these are also constructed during the initialization
period. For more information, see the State restoration during workload
rebalance section).
@@ -154,13 +154,13 @@ Example:
}
});
-For more information about standby replicas, see [Standby
Replicas](config-streams.md#num-standby-replicas).
+For more information about standby replicas, see [Standby
Replicas](../config-streams#num-standby-replicas).
# Elastic scaling of your application
Kafka Streams makes your stream processing applications elastic and scalable.
You can add and remove processing capacity dynamically during application
runtime without any downtime or data loss. This makes your applications
resilient in the face of failures and for allows you to perform maintenance as
needed (e.g. rolling upgrades).
-For more information about this elasticity, see the [Parallelism
Model](../architecture.html#streams_architecture_tasks) section. Kafka Streams
leverages the Kafka group management functionality, which is built right into
the [Kafka wire protocol](https://cwiki.apache.org/confluence/x/uxvVAQ). It is
the foundation that enables the elasticity of Kafka Streams applications:
members of a group coordinate and collaborate jointly on the consumption and
processing of data in Kafka. Additionall [...]
+For more information about this elasticity, see the [Parallelism
Model](../../architecture#streams_architecture_tasks) section. Kafka Streams
leverages the Kafka group management functionality, which is built right into
the [Kafka wire protocol](https://cwiki.apache.org/confluence/x/uxvVAQ). It is
the foundation that enables the elasticity of Kafka Streams applications:
members of a group coordinate and collaborate jointly on the consumption and
processing of data in Kafka. Additionally, [...]
## Adding capacity to your application
@@ -191,9 +191,9 @@ When a task is migrated, the task processing state is fully
restored before the
-For more information, see [Standby
Replicas](config-streams.html#num-standby-replicas).
+For more information, see [Standby
Replicas](../config-streams#num-standby-replicas).
-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
[`acceptable.recovery.lag`](config-streams.html#acceptable-recovery-lag), if
one exists. This means that most of the time, a task migration will **not**
result in downtime for that task. It will re [...]
+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
[`acceptable.recovery.lag`](../config-streams#acceptable-recovery-lag), if one
exists. This means that most of the time, a task migration will **not** result
in downtime for that task. It will rema [...]
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.
@@ -214,7 +214,7 @@ To achieve balanced workload processing across application
instances and to prev
There are a few things you can do to reduce the likelihood of crashes and
failures of your Kafka Streams application.
- * Kafka Streams has a few configurations that can help with resilience in
the face of broker failures. They can be found in the [configuration
guide.](config-streams.html#recommended-configuration-parameters-for-resiliency)
+ * Kafka Streams has a few configurations that can help with resilience in
the face of broker failures. They can be found in the [configuration
guide.](../config-streams#recommended-configuration-parameters-for-resiliency)
* Ensure that your application is able to handle errors and failures. This
includes things like configuring the correct exception handlers to handle
errors such as authorization and deserialization errors, and using strategies
such as dead letter queues to handle "poison pill" records.
diff --git a/docs/streams/developer-guide/security.md
b/docs/streams/developer-guide/security.md
index 4b0c1f97e4b..90877913b67 100644
--- a/docs/streams/developer-guide/security.md
+++ b/docs/streams/developer-guide/security.md
@@ -48,7 +48,7 @@ For more information about the security features in Apache
Kafka, see [Kafka Sec
# Required ACL setting for secure Kafka clusters
-Kafka clusters can use ACLs to control access to resources (like the ability
to create topics), and for such clusters each client, including Kafka Streams,
is required to authenticate as a particular user in order to be authorized with
appropriate access. In particular, when Streams applications are run against a
secured Kafka cluster, the principal running the application must have the ACL
set so that the application has the permissions to create, read and write
[internal topics](manage [...]
+Kafka clusters can use ACLs to control access to resources (like the ability
to create topics), and for such clusters each client, including Kafka Streams,
is required to authenticate as a particular user in order to be authorized with
appropriate access. In particular, when Streams applications are run against a
secured Kafka cluster, the principal running the application must have the ACL
set so that the application has the permissions to create, read and write
[internal topics](../man [...]
If the [streams rebalance
protocol](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol)
is enabled by setting `group.protocol=streams`, the following ACLs are
required on the topic and group resources:
diff --git a/docs/streams/developer-guide/testing.md
b/docs/streams/developer-guide/testing.md
index 25bb92c722c..4de4efa0422 100644
--- a/docs/streams/developer-guide/testing.md
+++ b/docs/streams/developer-guide/testing.md
@@ -230,7 +230,7 @@ The following example demonstrates how to use the test
driver and helper classes
# Unit Testing Processors
-If you [write a Processor](processor-api.html), you will want to test it.
+If you [write a Processor](../processor-api), you will want to test it.
Because the `Processor` forwards its results to the context rather than
returning them, Unit testing requires a mocked context capable of capturing
forwarded data for inspection. For this reason, we provide a
`MockProcessorContext` in `test-utils`.
@@ -320,7 +320,7 @@ Processors can schedule punctuators to handle periodic
tasks. The mock context d
final Punctuator punctuator = capturedPunctuator.getPunctuator();
punctuator.punctuate(/*timestamp*/ 0L);
-If you need to write tests involving automatic firing of scheduled
punctuators, we recommend creating a simple topology with your processor and
using the [`TopologyTestDriver`](testing.html#testing-topologytestdriver).
+If you need to write tests involving automatic firing of scheduled
punctuators, we recommend creating a simple topology with your processor and
using the [`TopologyTestDriver`](.#testing-topologytestdriver).
* [Documentation](/documentation)
* [Kafka Streams](/documentation/streams)
diff --git a/docs/streams/developer-guide/write-streams-app.md
b/docs/streams/developer-guide/write-streams-app.md
index 7f2a13dbfef..bb43c47cd0d 100644
--- a/docs/streams/developer-guide/write-streams-app.md
+++ b/docs/streams/developer-guide/write-streams-app.md
@@ -29,13 +29,13 @@ type: docs
-Any Java or Scala application that makes use of the Kafka Streams library is
considered a Kafka Streams application. The computational logic of a Kafka
Streams application is defined as a [processor
topology](../core-concepts#streams_topology), which is a graph of stream
processors (nodes) and streams (edges).
+Any Java or Scala application that makes use of the Kafka Streams library is
considered a Kafka Streams application. The computational logic of a Kafka
Streams application is defined as a [processor
topology](../../core-concepts#streams_topology), which is a graph of stream
processors (nodes) and streams (edges).
You can define the processor topology with the Kafka Streams APIs:
-[Kafka Streams DSL](dsl-api.html#streams-developer-guide-dsl)
- A high-level API that provides the most common data transformation
operations such as `map`, `filter`, `join`, and `aggregations` out of the box.
The DSL is the recommended starting point for developers new to Kafka Streams,
and should cover many use cases and stream processing needs. If you're writing
a Scala application then you can use the [Kafka Streams DSL for
Scala](dsl-api.html#scala-dsl) library which removes much of the Java/Scala
interoperability boilerplate as opposed to w [...]
-[Processor API](processor-api.html#streams-developer-guide-processor-api)
+[Kafka Streams DSL](../dsl-api#streams-developer-guide-dsl)
+ A high-level API that provides the most common data transformation
operations such as `map`, `filter`, `join`, and `aggregations` out of the box.
The DSL is the recommended starting point for developers new to Kafka Streams,
and should cover many use cases and stream processing needs. If you're writing
a Scala application then you can use the [Kafka Streams DSL for
Scala](../dsl-api#scala-dsl) library which removes much of the Java/Scala
interoperability boilerplate as opposed to wor [...]
+[Processor API](../processor-api#streams-developer-guide-processor-api)
A low-level API that lets you add and connect processors as well as
interact directly with state stores. The Processor API provides you with even
more flexibility than the DSL but at the expense of requiring more manual work
on the side of the application developer (e.g., more lines of code).
# Libraries and Maven artifacts
@@ -116,7 +116,7 @@ Description
**Tip**
-See the section [Data Types and
Serialization](datatypes.html#streams-developer-guide-serdes) for more
information about Serializers/Deserializers.
+See the section [Data Types and
Serialization](../datatypes#streams-developer-guide-serdes) for more
information about Serializers/Deserializers.
Example `pom.xml` snippet when using Maven:
@@ -143,7 +143,7 @@ You can call Kafka Streams from anywhere in your
application code, but usually t
First, you must create an instance of `KafkaStreams`.
- * The first argument of the `KafkaStreams` constructor takes a topology
(either `StreamsBuilder#build()` for the
[DSL](dsl-api.html#streams-developer-guide-dsl) or `Topology` for the
[Processor API](processor-api.html#streams-developer-guide-processor-api)) that
is used to define a topology.
+ * The first argument of the `KafkaStreams` constructor takes a topology
(either `StreamsBuilder#build()` for the
[DSL](../dsl-api#streams-developer-guide-dsl) or `Topology` for the [Processor
API](../processor-api#streams-developer-guide-processor-api)) that is used to
define a topology.
* The second argument is an instance of `java.util.Properties`, which
defines the configuration for this specific topology.
@@ -180,7 +180,7 @@ At this point, internal structures are initialized, but the
processing is not st
// Start the Kafka Streams threads
streams.start();
-If there are other instances of this stream processing application running
elsewhere (e.g., on another machine), Kafka Streams transparently re-assigns
tasks from the existing instances to the new instance that you just started.
For more information, see [Stream Partitions and
Tasks](../architecture.html#streams_architecture_tasks) and [Threading
Model](../architecture.html#streams_architecture_threads).
+If there are other instances of this stream processing application running
elsewhere (e.g., on another machine), Kafka Streams transparently re-assigns
tasks from the existing instances to the new instance that you just started.
For more information, see [Stream Partitions and
Tasks](../../architecture#streams_architecture_tasks) and [Threading
Model](../../architecture#streams_architecture_threads).
To catch any unexpected exceptions, you can set an
`java.lang.Thread.UncaughtExceptionHandler` before you start the application.
This handler is called whenever a stream thread is terminated by an unexpected
exception:
@@ -209,7 +209,7 @@ After an application is stopped, Kafka Streams will migrate
any tasks that had b
# Testing a Streams application
-Kafka Streams comes with a `test-utils` module to help you test your
application [here](testing.html).
+Kafka Streams comes with a `test-utils` module to help you test your
application [here](../testing).
* [Documentation](/documentation)
* [Kafka Streams](/documentation/streams)
