From adcac0b30a2261cf4c907ef7df329a800e7bd160 Mon Sep 17 00:00:00 2001 From: samarth-ksolves2024 Date: Wed, 27 Aug 2025 11:48:42 +0530 Subject: [PATCH 1/4] Updated running-app.html file for Available Listeners --- docs/streams/developer-guide/running-app.html | 75 +++++++++++++++++-- 1 file changed, 68 insertions(+), 7 deletions(-) diff --git a/docs/streams/developer-guide/running-app.html b/docs/streams/developer-guide/running-app.html index a6c603f2a3c21..cc79b03869910 100644 --- a/docs/streams/developer-guide/running-app.html +++ b/docs/streams/developer-guide/running-app.html @@ -134,17 +134,78 @@

Determining how many application instance
  • Data should be equally distributed across topic partitions. For example, if two topic partitions each have 1 million messages, this is better than a single partition with 2 million messages and none in the other.
  • Processing workload should be equally distributed across topic partitions. For example, if the time to process messages varies widely, then it is better to spread the processing-intensive messages across partitions rather than storing these messages within the same partition.
    • - - - + +

    Available Listeners and Callbacks

    +

    + Kafka Streams provides several listener and callback APIs that allow applications + to react to lifecycle events, handle errors, and observe state changes. +

    - - - + +     + From 82f936368a04ea6e148c938d893c0941380d45dc Mon Sep 17 00:00:00 2001 From: samarth-ksolves2024 Date: Tue, 30 Sep 2025 18:44:55 +0530 Subject: [PATCH 2/4] Fixed order of arguments to assertEquals in tools module --- .../java/org/apache/kafka/tools/ConfigCommandTest.java | 8 ++++---- .../org/apache/kafka/tools/LeaderElectionCommandTest.java | 4 ++-- .../consumer/group/ResetConsumerGroupOffsetTest.java | 2 +- .../kafka/tools/streams/StreamsGroupCommandTest.java | 4 ++-- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/tools/src/test/java/org/apache/kafka/tools/ConfigCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/ConfigCommandTest.java index 0aecba43463d1..13cfb28111132 100644 --- a/tools/src/test/java/org/apache/kafka/tools/ConfigCommandTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/ConfigCommandTest.java @@ -424,8 +424,8 @@ public void testParseConfigsToBeAddedForAddConfigFile() throws IOException { public void testExpectedEntityTypeNames(List expectedTypes, List expectedNames, List connectOpts, String... args) { ConfigCommand.ConfigCommandOptions createOpts = new ConfigCommand.ConfigCommandOptions(toArray(List.of(connectOpts.get(0), connectOpts.get(1), "--describe"), List.of(args))); createOpts.checkArgs(); - assertEquals(createOpts.entityTypes().toSeq(), seq(expectedTypes)); - assertEquals(createOpts.entityNames().toSeq(), seq(expectedNames)); + assertEquals(seq(expectedTypes), createOpts.entityTypes().toSeq()); + assertEquals(seq(expectedNames), createOpts.entityNames().toSeq()); } @Test @@ -1177,8 +1177,8 @@ public void shouldNotUpdateConfigIfNonExistingConfigIsDeleted() { public synchronized DescribeConfigsResult describeConfigs(Collection resources, DescribeConfigsOptions options) { assertEquals(1, resources.size()); ConfigResource res = resources.iterator().next(); - assertEquals(res.type(), ConfigResource.Type.TOPIC); - assertEquals(res.name(), resourceName); + assertEquals(ConfigResource.Type.TOPIC, res.type()); + assertEquals(resourceName, res.name()); return describeResult; } }; diff --git a/tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java index 2e3ebf6a7bd08..8809e3ca174b1 100644 --- a/tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/LeaderElectionCommandTest.java @@ -124,8 +124,8 @@ public void testDeprecatedAdminConfig() throws Exception { mockedAdmin.verify(() -> Admin.create(argumentCaptor.capture())); // verify that properties provided to admin client are the overridden properties final Properties actualProps = argumentCaptor.getValue(); - assertEquals(actualProps.get(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG), requestTimeoutMs); - assertEquals(actualProps.get(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG), defaultApiTimeoutMs); + assertEquals(requestTimeoutMs, actualProps.get(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG)); + assertEquals(defaultApiTimeoutMs, actualProps.get(AdminClientConfig.DEFAULT_API_TIMEOUT_MS_CONFIG)); } } diff --git a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java index bbcfb6e35c16a..503a5bea217e9 100644 --- a/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/consumer/group/ResetConsumerGroupOffsetTest.java @@ -909,7 +909,7 @@ private void resetAndAssertOffsetsCommitted(ClusterInstance cluster, Map> allResetOffsets = resetOffsets(service); allResetOffsets.forEach((group, offsetsInfo) -> offsetsInfo.forEach((tp, offsetMetadata) -> { - assertEquals(offsetMetadata.offset(), expectedOffsets.get(tp)); + assertEquals(expectedOffsets.get(tp), offsetMetadata.offset()); assertEquals(expectedOffsets, committedOffsets(cluster, topic, group)); })); } diff --git a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java index 4f1e116437e63..9ec961ebe77cc 100644 --- a/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/streams/StreamsGroupCommandTest.java @@ -504,9 +504,9 @@ private static void assertThrow(final String wrongState) { final String[] exceptionMessage = exception.getMessage().split(" Valid states are: "); assertEquals("Invalid state list '" + wrongState + "'.", exceptionMessage[0]); - assertEquals(Arrays.stream(exceptionMessage[1].split(",")) + assertEquals(validStates, Arrays.stream(exceptionMessage[1].split(",")) .map(String::trim) - .collect(Collectors.toSet()), validStates); + .collect(Collectors.toSet())); } private DescribeStreamsGroupsResult describeStreamsResult(String groupId, GroupState groupState) { From fab52c4c64ec8635997571ace80cad381656b18c Mon Sep 17 00:00:00 2001 From: samarth-ksolves2024 Date: Fri, 3 Oct 2025 15:15:58 +0530 Subject: [PATCH 3/4] Reverted running-app.html changes --- docs/streams/developer-guide/running-app.html | 383 ++++++++---------- 1 file changed, 161 insertions(+), 222 deletions(-) diff --git a/docs/streams/developer-guide/running-app.html b/docs/streams/developer-guide/running-app.html index cc79b03869910..bc6fab36bd4a1 100644 --- a/docs/streams/developer-guide/running-app.html +++ b/docs/streams/developer-guide/running-app.html @@ -18,232 +18,171 @@ - - - -
    - -
    - -
      -
    • Documentation
      • -
    • Kafka Streams
      • -
    • Developer Guide
      • +
      +

      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 section. Kafka Streams + leverages the Kafka group management functionality, which is built right into the Kafka wire protocol. 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

      +

      If you need more processing capacity for your stream processing application, you can simply start another instance of your stream processing application, e.g. on another machine, in order to scale out. The instances of your application will become aware of each other and automatically begin to share the processing work. More specifically, what will be handed over from the existing instances to the new instances is (some of) the stream tasks that have been run by the existing instances. Moving stream tasks from one instance to another results in moving the processing work plus any internal state of these stream tasks (the state of a stream task will be re-created in the target instance by restoring the state from its corresponding changelog topic).

      +

      The various instances of your application each run in their own JVM process, which means that each instance can leverage all the processing capacity that is available to their respective JVM process (minus the capacity that any non-Kafka-Streams part of your application may be using). This explains why running additional instances will grant your application additional processing capacity. The exact capacity you will be adding by running a new instance depends of course on the environment in which the new instance runs: available CPU cores, available main memory and Java heap space, local storage, network bandwidth, and so on. Similarly, if you stop any of the running instances of your application, then you are removing and freeing up the respective processing capacity.

      +
      + +

      Before adding capacity: only a single instance of your Kafka Streams application is running. At this point the corresponding Kafka consumer group of your application contains only a single member (this instance). All data is being read and processed by this single instance.

      +
      +
      + +

      After adding capacity: now two additional instances of your Kafka Streams application are running, and they have automatically joined the application’s Kafka consumer group for a total of three current members. These three instances are automatically splitting the processing work between each other. The splitting is based on the Kafka topic partitions from which data is being read.

      +
      +
      +
      +

      Removing capacity from your application

      +

      To remove processing capacity, you can stop running stream processing application instances (e.g., shut down two of + the four instances), it will automatically leave the application’s consumer group, and the remaining instances of + your application will automatically take over the processing work. The remaining instances take over the stream tasks that + were run by the stopped instances. Moving stream tasks from one instance to another results in moving the processing + work plus any internal state of these stream tasks. The state of a stream task is recreated in the target instance + from its changelog topic.

      +
      + +
      +
      +
      +

      State restoration during workload rebalance

      +

      When a task is migrated, the task processing state is fully restored before the application instance resumes + processing. This guarantees the correct processing results. In Kafka Streams, state restoration is usually done by + replaying the corresponding changelog topic to reconstruct the state store. To minimize changelog-based restoration + latency by using replicated local state stores, you can specify num.standby.replicas. When a stream task is + initialized or re-initialized on the application instance, its state store is restored like this:

      +
        +
      • If no local state store exists, the changelog is replayed from the earliest to the current offset. This reconstructs the local state store to the most recent snapshot.
        • +
      • If a local state store exists, the changelog is replayed from the previously checkpointed offset. The changes are applied and the state is restored to the most recent snapshot. This method takes less time because it is applying a smaller portion of the changelog.
      -
      -
      -
      - - \ No newline at end of file +
    +
    + + + + + +
    + +
    + + +
    +
    +
    + + \ No newline at end of file From ef12c30795a572d0a934667cb592c67ef3ba63ae Mon Sep 17 00:00:00 2001 From: samarth-ksolves2024 Date: Fri, 3 Oct 2025 15:36:11 +0530 Subject: [PATCH 4/4] Reverted running-app.html --- docs/streams/developer-guide/running-app.html | 316 +++++++++--------- 1 file changed, 158 insertions(+), 158 deletions(-) diff --git a/docs/streams/developer-guide/running-app.html b/docs/streams/developer-guide/running-app.html index bc6fab36bd4a1..a6c603f2a3c21 100644 --- a/docs/streams/developer-guide/running-app.html +++ b/docs/streams/developer-guide/running-app.html @@ -18,171 +18,171 @@ + + + + - - -
    - -
    - - -
    -
    -
    - - \ No newline at end of file + // Display docs subnav items + $('.b-nav__docs').parent().toggleClass('nav__item__with__subs--expanded'); + }); + \ No newline at end of file