KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies #12161

jnh5y · 2022-05-13T17:31:27Z

This PR adds the ability to pause and resume KafkaStreams instances as well as named/modular topologies.

Added an integration test to show how pausing and resuming works.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

wcarlson5

looking good for the most part just had a couple questions.

I gave the tests a quick pass. I realize you are still working on them so feel free to ignore that part until you are ready

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

This PR adds the ability to pause and resume KafkaStreams instances as well as named/modular topologies.

jnh5y · 2022-05-17T17:40:24Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        kafkaStreams.pause();
+        kafkaStreams.start();


I think the names here are a little confusing. With context, it makes sense that this is how to start a KafkaStreams instance with processing paused.

If anyone has a naming suggestion here, I'm very open to it!

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

cadonna

Thanks for the PR @jnh5y !

I would like to see a bit more unit testing.

I was wondering whether the pausing of tasks in restoration really works. See my comment in the StoreChangelogReader.

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

cadonna · 2022-05-18T09:52:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -897,7 +897,8 @@ private void initializeAndRestorePhase() {
        }
        // we can always let changelog reader try restoring in order to initialize the changelogs;
        // if there's no active restoring or standby updating it would not try to fetch any data
-        changelogReader.restore(taskManager.tasks());
+        // After KAFKA-13873, we only restore the not paused tasks.
+        changelogReader.restore(taskManager.notPausedTasks());


This should also be verified in a unit test with a mock changelog reader.

I think I've covered this with a new test in StreamThreadTest.

cadonna · 2022-05-18T09:56:18Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadata.java

        this.hasNamedTopologies = !(allTopologyNames.size() == 1 && allTopologyNames.contains(UNNAMED_TOPOLOGY));
+        this.pausedTopologies = pausedTopologies;
    }

    public boolean canProcessTask(final Task task, final long now) {


I am aware that there are now unit tests for this class, but there are enough different code paths that would justify to add unit tests for this method.

I've added some unit tests for TaskExecutionMetadata here: 0a37879

cadonna · 2022-05-18T10:01:03Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskExecutor.java

@@ -275,7 +275,7 @@ private void commitSuccessfullyProcessedTasks() {
    int punctuate() {
        int punctuated = 0;

-        for (final Task task : tasks.activeTasks()) {
+        for (final Task task : tasks.notPausedTasks()) {


Also here unit tests would be great and easily doable.

@cadonna Ok, I tried to add some unit tests for TaskExecutor and I think I was needing to mock quite a few things. Does that seem right? Or is there an easier way?

Note to self: Use notPausedActiveTasks.

cadonna · 2022-05-18T10:04:43Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

@@ -273,6 +274,12 @@ Collection<Task> allTasks() {
        return readOnlyTasks;
    }

+    Collection<Task> notPausedTasks() {


Unit tests would be great!

Unit tests would still be great!

@cadonna Same deal here, I started to try and test Tasks instances and I started down a path of mocking quite a bit of the internal classes.

I think I'm either missing an easier approach or testing these functions directly may require a decent amount of effort.

cadonna · 2022-05-18T10:07:08Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java

+     * Pauses a topology by name
+     * @param topologyName Name of the topology to pause
+     */
+    public void pauseTopology(final String topologyName) {


These methods could also be unit tested really easily.

Added tests here: 34ec8ac

…ored.

cadonna

Thanks for the updates, @jnh5y!

I did a quick pass. Here my feedback.

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

cadonna · 2022-05-24T13:46:03Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

+            try {
+                streams.cleanUp();
+                fail("Should have thrown IllegalStateException");
+            } catch (final IllegalStateException expected) {
+                assertEquals("Cannot clean up while running.", expected.getMessage());
+            }


Wouldn't it be simpler to use assertThrows()?

I'll admit to it; I just copy-pasted the test directly above:

kafka/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

Lines 816 to 821 in 34ec8ac

try {

streams.cleanUp();

fail("Should have thrown IllegalStateException");

} catch (final IllegalStateException expected) {

assertEquals("Cannot clean up while running.", expected.getMessage());

}

Is following the existing codebase ok or shall I spend some time to clean it up?

I would use assertThrows() in the new test. The existing codebase also uses it in some tests like shouldThrowExceptionSettingUncaughtExceptionHandlerNotInCreateState.
I would open a new PR to refactor the tests that use try-fail-catch. That is optional.

Fixed in 759d3a8

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

cadonna · 2022-05-24T14:06:44Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

+        thread.runOnce();
+
+        assertEquals(10L, store1.approximateNumEntries());
+        assertEquals(4L, store2.approximateNumEntries());


Why 4 and not 5?

Good question. Short answer: I'm not certain. I copied the test here: https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java#L1684-L1762 since it is dealing with updating standby tasks.

My guess is that the method is actually returning an approximated number of entries as suggested. I tried to verify that, but I got a little lost in the unit test's complexity.

Ah, I see! Could you please extract common code (especially the setup code) of those two tests into a method.
We could consider to just check for greater 0.
Additionally to standbys, we should also test if active tasks in restoration are paused.

For the refactoring, I've taken a pass at it here: a3bd8ae.

I tend to prefer to add new things consistently with existing code, so I'm hesitating to change the check. I'm fine either way.

I'll add something for active tasks if that's the way we go.

cadonna · 2022-05-24T14:34:54Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadataTest.java

+        // One second after the error, task1 cannot process, task2 can.
+        Assert.assertFalse(metadata.canProcessTask(mockTask1, 1000));
+        Assert.assertTrue(metadata.canProcessTask(mockTask2, 1000));
+


Could you add a case where the time since the error is exactly the backoff time (i.e. 5000)?
I would prefer to put the times 1000, 5000, and 10000 into variables with meaningful names instead of adding comments.

Sure. I took a crack at updating the test; lemme know if you'd prefer different names.

Could you please use CONSTANT_BACKOFF_MS - 1 for the verifications on line 111 and 112? I think it makes the intent clearer.

Fixed in cfb18f5

cadonna · 2022-05-24T14:47:20Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

+                    // for restoring active and updating standby we may prefer different poll time
+                    // in order to make sure we call the main consumer#poll in time.
+                    // TODO: once we move ChangelogReader to a separate thread this may no longer be a concern


I think these comments should be moved before the call to restoreConsumer.poll(). BTW, this again confirms that inline comments are a poor way to document code, most of the times.

cadonna · 2022-05-24T14:50:08Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

-            for (final TopicPartition partition : polledRecords.partitions()) {
-                bufferChangelogRecords(restoringChangelogByPartition(partition), polledRecords.records(partition));
-            }
+                // JNH: Fix this?


Please do not forget to remove those comments here and elsewhere.

cadonna

@jnh5y Thank you for the updates!

Here my feedback:

cadonna · 2022-06-01T09:22:44Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadataTest.java

+        // One second after the error, task1 cannot process, task2 can.
+        Assert.assertFalse(metadata.canProcessTask(mockTask1, 1000));
+        Assert.assertTrue(metadata.canProcessTask(mockTask2, 1000));
+


Could you please use CONSTANT_BACKOFF_MS - 1 for the verifications on line 111 and 112? I think it makes the intent clearer.

cadonna · 2022-06-01T09:28:55Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

+            try {
+                streams.cleanUp();
+                fail("Should have thrown IllegalStateException");
+            } catch (final IllegalStateException expected) {
+                assertEquals("Cannot clean up while running.", expected.getMessage());
+            }


I would use assertThrows() in the new test. The existing codebase also uses it in some tests like shouldThrowExceptionSettingUncaughtExceptionHandlerNotInCreateState.
I would open a new PR to refactor the tests that use try-fail-catch. That is optional.

cadonna · 2022-06-01T09:30:29Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java

@@ -273,6 +274,12 @@ Collection<Task> allTasks() {
        return readOnlyTasks;
    }

+    Collection<Task> notPausedTasks() {


Unit tests would still be great!

cadonna · 2022-06-01T09:41:09Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

+        thread.runOnce();
+
+        assertEquals(10L, store1.approximateNumEntries());
+        assertEquals(4L, store2.approximateNumEntries());


Ah, I see! Could you please extract common code (especially the setup code) of those two tests into a method.
We could consider to just check for greater 0.
Additionally to standbys, we should also test if active tasks in restoration are paused.

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

cadonna · 2022-06-01T11:11:43Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        // Verify no output somehow?
+        // Is there a better way to show this?
+        assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 0),
+            CoreMatchers.equalTo(Collections.emptyList()));


As you write in the comments, this is not straightforward. I think it is not good to just wait for no outputs for two reasons:

it increases the duration of the test unnecessarily

it opens the door for flakiness because we only rely on a duration for verification

I have the following idea. You create two identical KafkaStreams clients (be careful to specify distinct state directories for each). Both clients read from the same input topics but write to distinct output topics. You do the initial setup verifications for both. Then you pause one of the clients and wait until the other client produced a given amount of records to the output topic. When that happened, you verify if the paused client has not written anything to its output topic. The assumption is that both clients produce with the same rate to their output topics if they are not paused. If one of the two is paused and does not produce any records to the output topic in the same time the other client produces a certain number of records we can assume the pausing works.

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

cadonna · 2022-06-01T11:17:50Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+
+        // Verify that consumers read new data -- AKA, there is no lag.
+        final Map<String, Map<Integer, LagInfo>> lagMap = kafkaStreams.allLocalStorePartitionLags();
+        assertNoLag(lagMap);


Just to be clear, this works as long as the input buffers for polled records are not full. I think it is good enough for this test, though.

Ayup, that's what I understand and it is showing that the active task consumers are still reading (up until their buffers are full).

cadonna · 2022-06-01T11:30:04Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+
+
+    @Test
+    public void shouldPauseAndResumeKafkaStreams() throws Exception {


Could you please try to extract common code to reusable methods?

Yes. I've extracted some reusable methods.

cadonna · 2022-06-01T12:14:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

@@ -479,6 +485,26 @@ public void restore(final Map<TaskId, Task> tasks) {
        }
    }

+    private void updateStandbyPartitions(final Map<TaskId, Task> tasks,


I think you forgot to also pause the active tasks in restoration.

I'm wondering if we can make this a more general rule, like:

When reader is in Restore_Active state i.e. there are at least one (for sake of simplicity just say we happen to have exactly one) active task which needs restoration, say taskA; then taskA was paused, we should be able to transit to Update_Standby.

When reader is in Update_Standby state, and there is one active task say taskA resumed; we should be able to transit to Restore_Active.

I know for now it does not matter since we always pause all tasks with the current APIs, but this is extensible for finer-grained controls in the future.

cadonna · 2022-06-01T12:16:20Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

+                        restoreConsumer.resume(Collections.singleton(partition));
+                    } else {
+                        restoreConsumer.pause(Collections.singleton(partition));
+                    }


I think a better way would be to collect the partitions to resume and pause and then call restoreConsumer.pause() and restoreConsumer.restore() just once with the collections of partitions.

Done in cb6d6f1.

…seResumeIntegrationTest.java Co-authored-by: Bruno Cadonna <[email protected]>

…IMEOUT_MS_CONFIG.

jnh5y · 2022-06-03T00:56:17Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+    @Test
+    public void shouldWorkAcrossInstances() throws Exception {
+        // Create data on input topics (would need at least two partitions)
+        produceToInputTopics(INPUT_STREAM_1, STANDARD_INPUT_DATA);
+
+        // Start two instances paused
+        // Create KafkaStreams instance
+        final StreamsBuilder builder = new StreamsBuilder();
+        builder.stream(INPUT_STREAM_1).groupByKey().count().toStream().to(OUTPUT_STREAM_1);
+
+        kafkaStreams = new KafkaStreams(builder.build(props()), props());
+
+        // Start KafkaStream with paused processing.
+        kafkaStreams.pause();
+        kafkaStreams.start();
+        // Check for rebalancing instead?
+        waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, STARTUP_TIMEOUT);
+        assertTrue(kafkaStreams.isPaused());
+
+        // Create KafkaStreams instance
+        final StreamsBuilder builder2 = new StreamsBuilder();
+        builder2.stream(INPUT_STREAM_1).groupByKey().count().toStream().to(OUTPUT_STREAM_2);
+
+        kafkaStreams2 = new KafkaStreams(builder2.build(props()), props());
+
+        // Start KafkaStream with paused processing.
+        kafkaStreams2.pause();
+        kafkaStreams2.start();
+        // Check for rebalancing instead?
+        //waitForApplicationState(singletonList(kafkaStreams2), State.RUNNING, STARTUP_TIMEOUT);
+        assertTrue(kafkaStreams2.isPaused());
+
+        // Verify no data?
+        assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 0),
+            CoreMatchers.equalTo(Collections.emptyList()));
+
+        // -- Verify that each instance is in rebalancing (after change to pause active task restoration)
+
+        System.out.println("JNH: calling close: " + kafkaStreams.state());
+        // Close the other -- this causes a rebalance
+        kafkaStreams2.close();
+        waitForApplicationState(singletonList(kafkaStreams2), State.NOT_RUNNING, STARTUP_TIMEOUT);
+
+        System.out.println("JNH: called close: " + kafkaStreams.state());
+
+        // Resume paused instance
+        kafkaStreams.resume();
+        System.out.println("JNH: called resume " + kafkaStreams.state());
+        waitForApplicationState(singletonList(kafkaStreams), State.RUNNING, STARTUP_TIMEOUT);
+        System.out.println("JNH: streams is running again " + new Date());
+
+        // Observe all data processed
+        assertThat(waitUntilMinKeyValueRecordsReceived(consumerConfig, OUTPUT_STREAM_1, 3),
+            CoreMatchers.equalTo(COUNT_OUTPUT_DATA));
+    }


@cadonna I added this as a test to show what happens between multiple clients. I am noticing that this test was taking 45+ seconds to run.

It looks like it is hitting the ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG and I'm trying to think through why that may be happening. Any ideas?

…aster.

Test clean up.

cadonna

@jnh5y Thank you for the updates!

Here my feedback!

...ms/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadataTest.java

cadonna · 2022-06-09T07:26:36Z

...ms/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadataTest.java

+    @Test
+    public void testCanProcessWithoutNamedTopologies() {
+        final Set<String> topologies = Collections.singleton(UNNAMED_TOPOLOGY);
+        final Set<String> pausedTopologies = ConcurrentHashMap.newKeySet();


Out of curiosity, why do you use a ConcurrentHashMap here?

I used ConcurrentHashMap since that's what pausedTopologies is created as elsewhere.

In the unit tests, it can be a HashSet, I switched to that.

...ms/src/test/java/org/apache/kafka/streams/processor/internals/TaskExecutionMetadataTest.java

cadonna · 2022-06-09T07:39:53Z

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

-import java.io.File;
-import java.time.Duration;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Properties;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicLong;
-import java.util.stream.Stream;
-
-import static java.util.Collections.emptyMap;
-import static java.util.Collections.emptySet;
-import static java.util.Collections.singleton;
-import static java.util.Collections.singletonMap;
-import static org.apache.kafka.common.utils.Utils.mkEntry;
-import static org.apache.kafka.common.utils.Utils.mkMap;
-import static org.apache.kafka.common.utils.Utils.mkProperties;
-import static org.apache.kafka.common.utils.Utils.mkSet;
-import static org.apache.kafka.streams.processor.internals.ClientUtils.getSharedAdminClientId;
-import static org.apache.kafka.streams.processor.internals.StateManagerUtil.CHECKPOINT_FILE_NAME;
-import static org.easymock.EasyMock.anyObject;
-import static org.easymock.EasyMock.anyInt;
-import static org.easymock.EasyMock.expect;
-import static org.easymock.EasyMock.expectLastCall;
-import static org.easymock.EasyMock.mock;
-import static org.easymock.EasyMock.niceMock;
-import static org.easymock.EasyMock.verify;
-import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.CoreMatchers.not;
-import static org.hamcrest.CoreMatchers.startsWith;
-import static org.hamcrest.MatcherAssert.assertThat;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.is;
-import static org.hamcrest.Matchers.isA;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertNotNull;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertSame;
-import static org.junit.Assert.assertThrows;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-


I think you IDE reformatted the imports. The same occurred in TaskExecutionMetadataTest. Could you check also in other files? We use more or less the following import order:

all other imports non-static (sorted alphabetically)

java.* (sorted alphabetically)
javax.* (sorted alphabetically)

static imports (sorted alphabetically)

However, we are not really consistent across files. Nevertheless we should try to keep that order.

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

cadonna · 2022-06-09T08:31:22Z

checkstyle/suppressions.xml

@@ -225,7 +225,7 @@
              files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|KTableKTableForeignKeyJoinIntegrationTest|RocksDBGenericOptionsToDbOptionsColumnFamilyOptionsAdapterTest|RelationalSmokeTest|MockProcessorContextStateStoreTest).java"/>

    <suppress checks="JavaNCSS"
-              files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|TaskManagerTest).java"/>
+              files="(EosV2UpgradeIntegrationTest|KStreamKStreamJoinTest|StreamThreadTest|TaskManagerTest).java"/>


In general, I think we should try to avoid adding suppressions. But I also see that StreamThreadTest would need quite some love at the moment which is not the intent of this PR.

cadonna · 2022-06-09T09:09:10Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        produceToInputTopics(INPUT_STREAM_1, STANDARD_INPUT_DATA);
+        assertNoLag(kafkaStreams);
+
+        waitUntilStreamsHasPolled(kafkaStreams, 2);


I like your approach!

Thanks! It was either going to be a good idea or prove too hacky! Glad you like it!

It is hacky, but still a good idea 🙂

cadonna · 2022-06-09T09:21:28Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        assertTrue(kafkaStreams.isPaused());
+
+        produceToInputTopics(INPUT_STREAM_1, STANDARD_INPUT_DATA);
+        assertNoLag(kafkaStreams);


I just realized that this method computes the lag of the store regarding the changelog partition and not of the input partitions. Was this intentional? I think the lag of the store regarding the changelog topic will always be zero in this case because in normal mode the Streams client writes to the changelog topic and so it is always up-to-date.
I thought that this method verified that the main consumer does not have any lag regarding the input partitions which would tell us that the main consumer polled data although it was paused which is the expected behavior. This method with waitUntilStreamsHasPolled(kafkaStreams, 2); and assertTopicSize(OUTPUT_STREAM_1, <same size as before the call to pause>); would tell us that although the Streams clients polled data and went through the poll loop, no data was produced to the output topic. For that you should also set the cache size to 0 with STATESTORE_CACHE_MAX_BYTES_CONFIG.

First, good catch! It wasn't intentional... I thought that I had found the right calls to show no lag for consumers...

I tried to switch over to verifying that there was no consumer lag, and I don't think we can sensibly do so.

(If I understand correctly) The consumer offsets are only committed after the input records have been processed. Since there is no processing, the consumer offsets will appear to be behind. (I verified this locally.)

My conclusion is that the method isn't checking anything useful, so I am planning on removing it.

I agree that you should remove assertNoLag().

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

cadonna · 2022-06-09T09:42:22Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+
+        awaitOutput(OUTPUT_STREAM_1, 3, COUNT_OUTPUT_DATA);
+    }
+


Could you please also add a test that verifies if active tasks in restoration and standbys are paused? Something like you start two Streams clients with 1 standby. If you have one partition one client should get the active stateful task and the other should get the standby task. The clients process some data and write some data into their states. Then shutdown the Streams clients and wipe out the local state. Finally, start both clients paused and verify if the lag of the local stores stays constant and greater than zero for a couple of poll loop iterations.

I like the idea, but based on https://github.com/apache/kafka/pull/12161/files#r893999817 I'm not sure if we can sensibly check something useful.

It seems like the offsets (at least for the main consumer) will not be committed, so they will be definitely be at 0 since processing is paused.

For the restore consumer, does it have a groupId? Would we be able to see anything about its state?

The comment https://github.com/apache/kafka/pull/12161/files#r893999817 is unrelated here. In my proposal, you would assert that the lag between local state store and changelog topic does not decrease. You can measure that in a similar way as you already do in assertNoLag().
The restore consumer does not have a group since the partitions are manually assigned to the consumer. If there is no group there are no committed offsets.

I'm still a little light on the details. I'll give it a try here in a little bit.

Ok, I added a test named pausedTopologyShouldNotRestoreStateStores that hopefully will cover things.

cadonna · 2022-06-09T11:06:43Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        kafkaStreams2 = buildKafkaStreams(OUTPUT_STREAM_2);
+        kafkaStreams2.pause();
+        kafkaStreams2.start();
+        assertTrue(kafkaStreams2.isPaused());


Do you also want to wait for RUNNING for the second KafkaStreams instance?

Yes, I think that's fair.

cadonna · 2022-06-09T11:09:09Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Long().getClass());
+        properties.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000L);
+        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+        properties.put(StreamsConfig.TOPOLOGY_OPTIMIZATION_CONFIG, StreamsConfig.OPTIMIZE);


I think you can remove this config. It is not relevant for what this test tests.

cadonna · 2022-06-09T11:18:55Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+    }
+
+    @Test
+    public void pauseResumehouldWorkAcrossInstances() throws Exception {


According to the builds, this test seems flaky. I think the reason is that the cache (in Streams, not the RocksDB cache) of the state stores is not set to zero. When the cache is not set to zero, the number of results that are send downstreams is not deterministic, because some intermediate results might be sent downstream and some not. I tried to set the cache to zero and was able to run 100+ times the test in a row without failure whereas with the cache > 0, the test failed much earlier.
Note that by changing the cache size also the results that you verify change.

Thanks. I was having trouble shifting through the build info. I've updated the test to set the state store size to 0.

I'll watch the CI jobs to see if they are green after my last push. (They seemed to be ok as I was working today.)

I think you fixed it! 🎉

…nals/TaskExecutionMetadataTest.java Co-authored-by: Bruno Cadonna <[email protected]>

…nals/StreamThreadTest.java Co-authored-by: Bruno Cadonna <[email protected]>

cadonna

@jnh5y Thanks you for the updates!

Here my feedback.

cadonna · 2022-06-15T10:04:56Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        assertTrue(kafkaStreams.allLocalStorePartitionLags().isEmpty());
+        assertTrue(kafkaStreams2.allLocalStorePartitionLags().isEmpty());


Why are you verifying for emptiness? I would expect that there are entries for the state stores with a lag greater than 0.

I played a bit around with the test and indeed if you add a Thread.sleep(2000) before these asserts, the test fails because the returned map is not empty. That means, the assignment was not finished before the asserts were called.

You could do something like:

waitForApplicationState(Arrays.asList(kafkaStreams), State.REBALANCING, STARTUP_TIMEOUT); waitForCondition( () -> !kafkaStreams.allLocalStorePartitionLags().isEmpty(), "Lags for local store partitions were not found within the timeout!"); waitUntilStreamsHasPolled(kafkaStreams, 2); final long stateStoreLag1 = kafkaStreams.allLocalStorePartitionLags().get("test-store").get(0).offsetLag(); waitUntilStreamsHasPolled(kafkaStreams, 2); final long stateStoreLag2 = kafkaStreams.allLocalStorePartitionLags().get("test-store").get(0).offsetLag(); assertTrue(stateStoreLag1 > 0); assertEquals(stateStoreLag1, stateStoreLag2);

This code just considers one Streams client. You need to add Materialized.as("test-store") to the call to count() in your topology.
As soon as you activated the standbys, you need to do the same for the second Streams client.

Thank you! I've added these changes.

cadonna · 2022-06-15T10:09:01Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        kafkaStreams = buildKafkaStreams(OUTPUT_STREAM_1);
+        kafkaStreams2 = buildKafkaStreams(OUTPUT_STREAM_1);


If you do not use standby tasks, there is no reason to use two Kafka Streams clients. I would propose to use one standby only for this test. For that you need to set num.standby.replicas to 1. That has the effect that one client gets the active store assigned and the other gets the standby store assigned.

My mistake; I've updated the test.

cadonna · 2022-06-15T10:13:15Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

@@ -479,6 +485,47 @@ public void restore(final Map<TaskId, Task> tasks) {
        }
    }

+    private void updateStandbyPartitions(final Map<TaskId, Task> tasks,


Do not forget to rename this method to something more meaningful.
Proposal: pauseResumePartitions()

cadonna

@jnh5y Thank you for the updates!

LGTM!

Had just one nit.

Thank you for your patience!

cadonna · 2022-06-15T11:55:29Z

streams/src/test/java/org/apache/kafka/streams/integration/PauseResumeIntegrationTest.java

+        assertStreamsLagStaysConstant(kafkaStreams);
+        assertStreamsLagStaysConstant(kafkaStreams2);


nit: assertStreamsLagStaysConstant() -> assertStreamsLocalStoreLagStaysConstant()

Changed it!

jnh5y · 2022-06-15T12:03:44Z

@jnh5y Thank you for the updates!

LGTM!

Had just one nit.

Thank you for your patience!

@cadonna Thank you for pushing me and helping me learn more about streams!

…ache#12161) This PR adds the ability to pause and resume KafkaStreams instances as well as named/modular topologies (KIP-834). Co-authored-by: Bruno Cadonna <[email protected]> Reviewers: Bonnie Varghese <[email protected]>, Walker Carlson <[email protected]>, Guozhang Wang <[email protected]>, Bruno Cadonna <[email protected]>

mjsax added streams kip Requires or implements a KIP labels May 13, 2022

wcarlson5 reviewed May 13, 2022

View reviewed changes

jnh5y force-pushed the kafka-13873 branch from 3fb63c3 to 0710747 Compare May 16, 2022 15:01

jnh5y marked this pull request as ready for review May 16, 2022 15:06

jnh5y changed the title ~~DRAFT: KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies~~ KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies May 16, 2022

KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies

207c4ef

This PR adds the ability to pause and resume KafkaStreams instances as well as named/modular topologies.

jnh5y force-pushed the kafka-13873 branch from 0710747 to 207c4ef Compare May 16, 2022 17:43

jnh5y commented May 17, 2022

View reviewed changes

bvarghese1 reviewed May 17, 2022

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java Outdated Show resolved Hide resolved

bvarghese1 reviewed May 17, 2022

View reviewed changes

streams/src/main/java/org/apache/kafka/streams/processor/internals/Tasks.java Outdated Show resolved Hide resolved

cadonna reviewed May 18, 2022

View reviewed changes

jnh5y added 8 commits May 18, 2022 08:32

Cleaning up reviewer comments.

b73708f

Some work on fixing up the StoreChangelogReader.

070b867

Updates to the IT.

d019d7a

IntelliJ refactoring to fix complexity.

66fc07e

Adding unit tests for KafkaStreams.

e0a80bb

Adding unit test showing that only non-paused, standby tasks are rest…

85e506d

…ored.

Found the suppressions for JavaNCSS.

a620621

Added TaskExecutionMetadataTest.

0a37879

jnh5y force-pushed the kafka-13873 branch from 3fec04b to 0a37879 Compare May 23, 2022 21:27

Adding TopologyMetadataTest.

34ec8ac

cadonna reviewed May 24, 2022

View reviewed changes

jnh5y added 3 commits May 24, 2022 16:00

Fixes to make the StoreChangelogReader work.

fb016a8

Responding to reviewer comments.

6d57d04

Minor refactoring / renaming.

d007d15

cadonna reviewed Jun 1, 2022

View reviewed changes

Update streams/src/test/java/org/apache/kafka/streams/integration/Pau…

d7be172

…seResumeIntegrationTest.java Co-authored-by: Bruno Cadonna <[email protected]>

jnh5y added 3 commits June 2, 2022 19:49

Test which shows that active tasks still restore.

148da23

Pausing active task restoration.

e467852

Updated PR IT with multiple client example. This is hitting SESSION_T…

1bcc5c9

…IMEOUT_MS_CONFIG.

jnh5y commented Jun 3, 2022

View reviewed changes

jnh5y added 5 commits June 3, 2022 11:49

Updating the PauseResume IT so that the multiple instance test runs f…

08e332d

…aster.

Fixing StoreChangelogReaderTest.

c495fb6

Removing most comments.

771c296

Added a little getTopicSize function.

e076a07

Verifying Streams pausing by watching the poll count.

f56dd99

Test clean up.

cadonna reviewed Jun 9, 2022

View reviewed changes

jnh5y and others added 9 commits June 9, 2022 08:40

Update streams/src/test/java/org/apache/kafka/streams/processor/inter…

6771854

…nals/TaskExecutionMetadataTest.java Co-authored-by: Bruno Cadonna <[email protected]>

Update streams/src/test/java/org/apache/kafka/streams/processor/inter…

12342d5

…nals/TaskExecutionMetadataTest.java Co-authored-by: Bruno Cadonna <[email protected]>

Update streams/src/test/java/org/apache/kafka/streams/processor/inter…

61a302b

…nals/StreamThreadTest.java Co-authored-by: Bruno Cadonna <[email protected]>

Updating import order and sorting out other issues.

06c98ea

Responding to more reviewer comments.

1cdbc15

Small test changes before adding new test.

ce707de

Partial unit test?

769a53a

Added new unit test.

ea77133

Removing assertNoLag.

f7799a5

cadonna reviewed Jun 15, 2022

View reviewed changes

Responding to reviewer comments.

e1c698b

cadonna approved these changes Jun 15, 2022

View reviewed changes

Responding to reviewer comments.

c5d7abc

jnh5y mentioned this pull request Jun 15, 2022

feat: Support pausing/resuming persistent queries confluentinc/ksql#9203

Merged

2 tasks

cadonna merged commit 7ed3748 into apache:trunk Jun 16, 2022

	try {
	streams.cleanUp();
	fail("Should have thrown IllegalStateException");
	} catch (final IllegalStateException expected) {
	assertEquals("Cannot clean up while running.", expected.getMessage());
	}



		@Test
		public void shouldPauseAndResumeKafkaStreams() throws Exception {

		assertTrue(kafkaStreams.allLocalStorePartitionLags().isEmpty());
		assertTrue(kafkaStreams2.allLocalStorePartitionLags().isEmpty());

		kafkaStreams = buildKafkaStreams(OUTPUT_STREAM_1);
		kafkaStreams2 = buildKafkaStreams(OUTPUT_STREAM_1);

		assertStreamsLagStaysConstant(kafkaStreams);
		assertStreamsLagStaysConstant(kafkaStreams2);

KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies #12161

KAFKA-13873 Add ability to Pause / Resume KafkaStreams Topologies #12161

Conversation

jnh5y commented May 13, 2022

Committer Checklist (excluded from commit message)

wcarlson5 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cadonna left a comment

Choose a reason for hiding this comment