From c164c47987f4c2a2fe8f7ad9a0c1e86ac8d0fa26 Mon Sep 17 00:00:00 2001 From: Ming-Yen Chung Date: Sat, 11 Jul 2026 12:33:41 +0800 Subject: [PATCH] MINOR: Fix broken relative links in docs/ Several cross-page links in the docs use relative paths that assume a flat single-page-HTML layout (e.g. foo.html#anchor) or omit a directory level, both pre-dating the move to per-page Hugo directories (foo.md -> /foo/). These resolve one level too shallow and 404 on the site. Includes the Kafka Connect quickstart link and the new Streams/ Connect security model cross-links. --- 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 +- .../streams/developer-guide/app-reset-tool.md | 2 +- .../streams/developer-guide/config-streams.md | 14 ++--- docs/streams/developer-guide/datatypes.md | 6 +- docs/streams/developer-guide/dsl-api.md | 56 +++++++++---------- .../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 +- .../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 f2594cdd3e89e..71c888f737223 100644 --- a/docs/getting-started/upgrade.md +++ b/docs/getting-started/upgrade.md @@ -150,7 +150,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.QueuedReplicaToDirAssignments`, `kafka.log.remote:type=RemoteStorageThreadPool.RemoteLogReaderTaskQueueSize`, and `kafka.log.remote:type=RemoteStorageThreadPool.RemoteLogReaderAvgIdlePercent`. For further details, please refer to [KIP-1100](https://cwiki.apache.org/confluence/x/6oqMEw). * 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 and will be removed in Kafka 5.0.0. For further details, please refer to [KIP-1188](https://cwiki.apache.org/confluence/x/2IkvFg). + * 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` policy is now deprecated and will be removed in Kafka 5.0.0. For further details, please refer to [KIP-1188](https://cwiki.apache.org/confluence/x/2IkvFg). * 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 3f46d54b0b1c6..8436962716b0a 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 613d7535cfe59..2ebfda953838f 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 API callers: anyone who knows a valid template string can usually retrieve the resolved value through the REST API. @@ -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 8508ce5c9a1df..7904c2368f55a 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 b7f96add11afe..9b9d7929d7ad1 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 d5d776e462b55..3249e2d8366c7 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 51bacbc20663a..7b5d481d8bc90 100644 --- a/docs/streams/developer-guide/config-streams.md +++ b/docs/streams/developer-guide/config-streams.md @@ -1183,7 +1183,7 @@ Added to a windows maintainMs to ensure data is not deleted from the log prematu -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.) @@ -1200,7 +1200,7 @@ Sets window size for the deserializer in order to calculate window end times. -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.) @@ -1364,7 +1364,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 @@ -1375,7 +1375,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} @@ -1476,7 +1476,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 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 [State](../architecture.html#streams_architecture_state) section. +> 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 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 [State](../../architecture#streams_architecture_state) section. > > 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. @@ -1486,7 +1486,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 @@ -1617,7 +1617,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 default for parameter `commit.interval.ms` changes to 100ms. Additionally, consumers are configured with `isolation.level="read_committed"` and producers are configured with `enable.idempotence=true` per default. Note that by default exactly-once processing requires a cluster of at least three brokers, which is the recommended setting for production. For development, you can change this configuration by adjusting broker setting `transaction.state.log.replication.factor` and `transaction.state.log.min.isr` to the number of brokers you want to use. For more details see [Processing Guarantees](../core-concepts#streams_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 default for parameter `commit.interval.ms` changes to 100ms. Additionally, consumers are configured with `isolation.level="read_committed"` and producers are configured with `enable.idempotence=true` per default. Note that by default exactly-once processing requires a cluster of at least three brokers, which is the recommended setting for production. For development, you can change this configuration by adjusting broker setting `transaction.state.log.replication.factor` and `transaction.state.log.min.isr` to the number of brokers you want to use. For more details see [Processing Guarantees](../../core-concepts#streams_processing_guarantee). > > 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 4e246cf7f1b0f..dd25b1c697df6 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 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-handler) section. +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) section. # 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 6f5fc67f555cc..6ed4e375784b1 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 _streams_ of customer-related events into an output _table_ that contains a continuously updated 360-degree view of your customers. + * 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 _streams_ of customer-related events into an output _table_ that contains a continuously updated 360-degree view of your customers. * 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 (record stream). -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#streams-developer-guide-interactive-queries-local-key-value-stores) with an optional `queryableStoreName` argument. +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-developer-guide-interactive-queries-local-key-value-stores) with an optional `queryableStoreName` argument. @@ -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 be [compacted](memory-mgmt.html#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka 1)` directly to `(kafka, 3)`. Typically, you should only disable record caches for testing or debugging purposes - under normal circumstances it is better to leave record caches enabled. +**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 be [compacted](../memory-mgmt#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka 1)` directly to `(kafka, 3)`. Typically, you should only disable record caches for testing or debugging purposes - under normal circumstances it is better to leave record caches enabled.
@@ -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 be [compacted](memory-mgmt.html#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka 1)` directly to `(kafka, 3)`. Typically, you should only disable record caches for testing or debugging purposes - under normal circumstances it is better to leave record caches enabled. +**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 be [compacted](../memory-mgmt#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka 1)` directly to `(kafka, 3)`. Typically, you should only disable record caches for testing or debugging purposes - under normal circumstances it is better to leave record caches enabled.
@@ -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 day and can be changed via `Materialized#withRetention()`. +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 day and can be changed via `Materialized#withRetention()`. 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 delivery semantics when talking to such external systems (e.g., to retry on delivery failure or to prevent message duplication). +**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 delivery semantics when talking to such external systems (e.g., to retry on delivery failure or to prevent message duplication). # 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 d3fed2f4a14b3..53d3100cf4080 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. ![](/43/images/streams-interactive-queries-03.png) @@ -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, Kafka Streams will keep track of the RPC endpoint information for every instance of an application, its state stores, and assigned stream partitions through instances of [StreamsMetadata](/{version}/javadoc/org/apache/kafka/streams/state/StreamsMetadata.html). +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, Kafka Streams will keep track of the RPC endpoint information for every instance of an application, its state stores, and assigned stream partitions through instances of [StreamsMetadata](/{version}/javadoc/org/apache/kafka/streams/state/StreamsMetadata.html). **Tip** diff --git a/docs/streams/developer-guide/manage-topics.md b/docs/streams/developer-guide/manage-topics.md index 1a16e7ab2297a..aac4ab745e456 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 c63f997d16e45..b79a7348ddde0 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` with the records `: , , , `. 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`. + * 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`. > * **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: `, , `. > * **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 ``. 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 5bfb25cec4a63..a35e4f6e43d89 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` 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(newValue)` would copy the key, timestamp, and headers from `inputRecord` while setting the output record's value to `newValue`. Note that this does not mutate `inputRecord`, but instead creates a shallow copy. Beware that this is only a shallow copy, so if you plan to mutate the key, value, or headers elsewhere in the program, you will want to create a deep copy of those fields yourself. -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 `TimestampExtractor`). When stream-time is used, `punctuate()` is triggered purely by data because stream-time is determined (and advanced forward) by the timestamps derived from the input data. When there is no new input data arriving, stream-time is not advanced and thus `punctuate()` is not called. +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 `TimestampExtractor`). When stream-time is used, `punctuate()` is triggered purely by data because stream-time is determined (and advanced forward) by the timestamps derived from the input data. When there is no new input data arriving, stream-time is not advanced and thus `punctuate()` is not called. 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 dashboard or a microservice implemented in Scala or Go. +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 or a microservice implemented in Scala or Go. 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](): 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]() when you need a persistent plain key-value store (no embedded record timestamp). * Use [persistentTimestampedKeyValueStore]() 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 example, if you experience machine failure, the state store and the application's state can be fully restored from its changelog. You can enable or disable this backup feature for a state store. +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 example, if you experience machine failure, the state store and the application's state can be fully restored from its changelog. You can enable or disable this backup feature for a state store. 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 high enough to accommodate a record with one key arriving out-of-order relative to a record for another key. -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 `ProcessorSupplier` should always generate a new instance each time `ProcessorSupplier#get()` gets called. -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 32310df11ade4..edc56543dc9bf 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. Additionally, Kafka Streams provides stateful processing and allows for fault-tolerant state in environments where application instances may come and go at any time. +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, Kafka Streams provides stateful processing and allows for fault-tolerant state in environments where application instances may come and go at any time. ## 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 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 [regularly probe](config-streams.html#probing-rebalance-interval-ms) for warmup tasks that have finished restoring and transition them to active tasks when ready. +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 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 [regularly probe](../config-streams#probing-rebalance-interval-ms) for warmup tasks that have finished restoring and transition them to active tasks when ready. 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 4b0c1f97e4bcb..90877913b67d3 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-topics.html#streams-developer-guide-topics-internal). +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-topics#streams-developer-guide-topics-internal). 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 25bb92c722c27..4de4efa042289 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 7f2a13dbfefee..bb43c47cd0d41 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 working directly with the Java DSL. -[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 working directly with the Java DSL. +[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)