From e06dfb1bbcbe8f315a73a0811b30565e9c2af791 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Mon, 16 Mar 2020 13:50:01 -0700 Subject: [PATCH 1/5] add fixes --- .../processor/internals/StandbyTask.java | 37 ++++++++----------- .../internals/StoreChangelogReader.java | 12 ++++-- .../processor/internals/StreamTask.java | 8 ++-- .../processor/internals/TaskManager.java | 2 +- .../internals/StoreChangelogReaderTest.java | 21 +++++++++++ 5 files changed, 50 insertions(+), 30 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index 4f4dda9cf2c47..950ff151190a2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -42,6 +42,7 @@ public class StandbyTask extends AbstractTask implements Task { private final Logger log; private final String logPrefix; private final Sensor closeTaskSensor; + private final boolean eosEnabled; private final InternalProcessorContext processorContext; private Map offsetSnapshotSinceLastCommit; @@ -71,6 +72,7 @@ public class StandbyTask extends AbstractTask implements Task { processorContext = new StandbyContextImpl(id, config, stateMgr, metrics); closeTaskSensor = ThreadMetrics.closeTaskSensor(Thread.currentThread().getName(), metrics); + this.eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); } @Override @@ -201,36 +203,27 @@ public void closeDirty() { private void close(final boolean clean, final Map checkpoint) { - if (state() == State.CREATED) { - // the task is created and not initialized, do nothing - closeTaskSensor.record(); - transitionTo(State.CLOSED); - return; - } - - if (state() == State.RUNNING) { + if (state() == State.CREATED || state() == State.RUNNING) { if (clean) { // since there's no written offsets we can checkpoint with empty map, // and the state current offset would be used to checkpoint stateMgr.checkpoint(Collections.emptyMap()); offsetSnapshotSinceLastCommit = new HashMap<>(stateMgr.changelogOffsets()); } - - executeAndMaybeSwallow(clean, () -> { - StateManagerUtil.closeStateManager( - log, - logPrefix, - clean, - false, + final boolean wipeStateStore = !clean && eosEnabled; + + executeAndMaybeSwallow(clean, + () -> StateManagerUtil.closeStateManager( + log, + logPrefix, + wipeStateStore, + false, stateMgr, stateDirectory, - TaskType.STANDBY); - }, - "state manager close", - log - ); - - // TODO: if EOS is enabled, we should wipe out the state stores like we did for StreamTask too + TaskType.STANDBY), + "state manager close", + log + ); } else { throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java index b5e4ae404e791..15260a174e5c1 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java @@ -798,10 +798,16 @@ public void remove(final Collection revokedChangelogs) { for (final TopicPartition partition : revokedChangelogs) { final ChangelogMetadata changelogMetadata = changelogs.remove(partition); - if (changelogMetadata.state() != ChangelogState.REGISTERED) { - revokedInitializedChangelogs.add(partition); + if (changelogMetadata != null) { + if (changelogMetadata.state() != ChangelogState.REGISTERED) { + revokedInitializedChangelogs.add(partition); + } + changelogMetadata.clear(); + } else { + log.debug("Changelog partition {} could not be found, " + + "it could be already cleaned up during the handling" + + "of task corruption and never restore again", partition); } - changelogMetadata.clear(); } removeChangelogsFromRestoreConsumer(revokedInitializedChangelogs); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index beb0fd9919c7e..e0062241d6023 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -78,7 +78,7 @@ public class StreamTask extends AbstractTask implements ProcessorNodePunctuator, // we want to abstract eos logic out of StreamTask, however // there's still an optimization that requires this info to be // leaked into this class, which is to checkpoint after committing if EOS is not enabled. - private final boolean eosDisabled; + private final boolean eosEnabled; private final long maxTaskIdleMs; private final int maxBufferedSize; @@ -120,7 +120,7 @@ public StreamTask(final TaskId id, this.time = time; this.recordCollector = recordCollector; - eosDisabled = !StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); + eosEnabled = !StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); final String threadId = Thread.currentThread().getName(); closeTaskSensor = ThreadMetrics.closeTaskSensor(threadId, streamsMetrics); @@ -326,7 +326,7 @@ public void postCommit() { commitNeeded = false; commitRequested = false; - if (eosDisabled) { + if (!eosEnabled) { stateMgr.checkpoint(checkpointableOffsets()); } @@ -487,7 +487,7 @@ private void close(final boolean clean, case SUSPENDED: // if EOS is enabled, we wipe out the whole state store for unclean close // since they are invalid to use anymore - final boolean wipeStateStore = !clean && !eosDisabled; + final boolean wipeStateStore = !clean && eosEnabled; // first close state manager (which is idempotent) then close the record collector (which could throw), // if the latter throws and we re-close dirty which would close the state manager again. diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java index 071c7cb3b6312..becf1439bcac5 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java @@ -333,7 +333,7 @@ boolean tryToCompleteRestoration() { // it is possible that if there are multiple threads within the instance that one thread // trying to grab the task from the other, while the other has not released the lock since // it did not participate in the rebalance. In this case we can just retry in the next iteration - log.debug("Could not initialize {} due to {}; will retry", task.id(), e); + log.debug("Could not initialize {} due to the following exception; will retry", task.id(), e); allRunning = false; } } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index e033535cd4d51..60407fed97dc1 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -30,6 +30,7 @@ import org.apache.kafka.streams.errors.StreamsException; import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.internals.ProcessorStateManager.StateStoreMetadata; +import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender; import org.apache.kafka.test.MockStateRestoreListener; import org.apache.kafka.test.StreamsTestUtils; import org.easymock.EasyMock; @@ -46,6 +47,7 @@ import java.util.Collection; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; @@ -943,6 +945,25 @@ public Map endOffsets(final Collection par assertEquals(kaboom, thrown.getCause()); } + @Test + public void shouldNotThrowOnUnknownRevokedPartition() { + LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); + LogCaptureAppender.setClassLoggerToDebug(changelogReader.getClass()); + + try { + changelogReader.remove(Collections.singletonList(new TopicPartition("unknown", 0))); + + assertEquals(Collections.singletonList( + "test-reader Changelog partition unknown-0 could not be found, " + + "it could be already cleaned up during the handling" + + "of task corruption and never restore again"), appender.getMessages() + + ); + } finally { + LogCaptureAppender.unregister(appender); + } + } + private void setupConsumer(final long messages, final TopicPartition topicPartition) { assignPartition(messages, topicPartition); addRecords(messages, topicPartition); From 65bba1270b878a42f0e1666b44ef8509e6513a04 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 19 Mar 2020 15:54:34 -0700 Subject: [PATCH 2/5] standby task fix --- .../processor/internals/StandbyTask.java | 25 +++++++++---------- .../internals/StoreChangelogReaderTest.java | 1 - 2 files changed, 12 insertions(+), 14 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index 950ff151190a2..a7d54a8b6becc 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -189,20 +189,19 @@ private void prepareClose(final boolean clean) { @Override public void closeClean(final Map checkpoint) { Objects.requireNonNull(checkpoint); - close(true, checkpoint); + close(true); log.info("Closed clean"); } @Override public void closeDirty() { - close(false, null); + close(false); log.info("Closed dirty"); } - private void close(final boolean clean, - final Map checkpoint) { + private void close(final boolean clean) { if (state() == State.CREATED || state() == State.RUNNING) { if (clean) { // since there's no written offsets we can checkpoint with empty map, @@ -212,18 +211,18 @@ private void close(final boolean clean, } final boolean wipeStateStore = !clean && eosEnabled; - executeAndMaybeSwallow(clean, - () -> StateManagerUtil.closeStateManager( - log, - logPrefix, - wipeStateStore, - false, + executeAndMaybeSwallow(clean, () -> + StateManagerUtil.closeStateManager( + log, + logPrefix, + wipeStateStore, + false, stateMgr, stateDirectory, TaskType.STANDBY), - "state manager close", - log - ); + "state manager close", + log + ); } else { throw new IllegalStateException("Illegal state " + state() + " while closing standby task " + id); } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index 60407fed97dc1..116a6f2d5a023 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -957,7 +957,6 @@ public void shouldNotThrowOnUnknownRevokedPartition() { "test-reader Changelog partition unknown-0 could not be found, " + "it could be already cleaned up during the handling" + "of task corruption and never restore again"), appender.getMessages() - ); } finally { LogCaptureAppender.unregister(appender); From e4b254ced498c71573c6171dc2634d0f0d303552 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 19 Mar 2020 17:00:49 -0700 Subject: [PATCH 3/5] unclean close test --- .../processor/internals/StandbyTask.java | 2 +- .../processor/internals/StandbyTaskTest.java | 81 ++++++++++++------- 2 files changed, 55 insertions(+), 28 deletions(-) diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index a7d54a8b6becc..6e7c346f148e5 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -215,8 +215,8 @@ private void close(final boolean clean) { StateManagerUtil.closeStateManager( log, logPrefix, + clean, wipeStateStore, - false, stateMgr, stateDirectory, TaskType.STANDBY), diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java index a12da5b45d311..34a1f5fcb8799 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StandbyTaskTest.java @@ -62,7 +62,6 @@ import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; - @RunWith(EasyMockRunner.class) public class StandbyTaskTest { @@ -213,29 +212,6 @@ public void shouldReturnStateManagerChangelogOffsets() { EasyMock.verify(stateManager); } - @Test - public void shouldDoNothingWithCreatedStateOnClose() { - stateManager.close(); - EasyMock.expectLastCall().andThrow(new AssertionError("Close should not be called")).anyTimes(); - stateManager.flush(); - EasyMock.expectLastCall().andThrow(new AssertionError("Flush should not be called")).anyTimes(); - stateManager.checkpoint(EasyMock.anyObject()); - EasyMock.expectLastCall().andThrow(new AssertionError("Checkpoint should not be called")).anyTimes(); - EasyMock.replay(stateManager); - final MetricName metricName = setupCloseTaskMetric(); - - task = createStandbyTask(); - final Map checkpoint = task.prepareCloseClean(); - task.closeClean(checkpoint); - - assertEquals(Task.State.CLOSED, task.state()); - - final double expectedCloseTaskMetric = 1.0; - verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName); - - EasyMock.verify(stateManager); - } - @Test public void shouldNotCommitAndThrowOnCloseDirty() { stateManager.close(); @@ -407,12 +383,63 @@ public void shouldThrowOnCloseCleanCheckpointError() { public void shouldThrowIfClosingOnIllegalState() { task = createStandbyTask(); - final Map checkpoint = task.prepareCloseClean(); - task.closeClean(checkpoint); + task.transitionTo(Task.State.RESTORING); - // close call are not idempotent since we are already in closed + // close calls are not idempotent since we are already in closed assertThrows(IllegalStateException.class, task::prepareCloseClean); assertThrows(IllegalStateException.class, task::prepareCloseDirty); + + task.transitionTo(Task.State.CLOSED); + } + + @Test + public void shouldCloseStateManagerOnTaskCreated() { + stateManager.close(); + EasyMock.expectLastCall(); + + EasyMock.replay(stateManager); + + final MetricName metricName = setupCloseTaskMetric(); + + task = createStandbyTask(); + + task.closeDirty(); + + final double expectedCloseTaskMetric = 1.0; + verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName); + + EasyMock.verify(stateManager); + + assertEquals(Task.State.CLOSED, task.state()); + } + + @Test + public void shouldDeleteStateDirOnTaskCreatedAndEOSUncleanClose() { + stateManager.close(); + EasyMock.expectLastCall(); + + EasyMock.expect(stateManager.baseDir()).andReturn(baseDir); + + EasyMock.replay(stateManager); + + final MetricName metricName = setupCloseTaskMetric(); + + config = new StreamsConfig(mkProperties(mkMap( + mkEntry(StreamsConfig.APPLICATION_ID_CONFIG, applicationId), + mkEntry(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:2171"), + mkEntry(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE) + ))); + + task = createStandbyTask(); + + task.closeDirty(); + + final double expectedCloseTaskMetric = 1.0; + verifyCloseTaskMetric(expectedCloseTaskMetric, streamsMetrics, metricName); + + EasyMock.verify(stateManager); + + assertEquals(Task.State.CLOSED, task.state()); } private StandbyTask createStandbyTask() { From 3245b2dad1504926206039d989f9c7ea05480a03 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 19 Mar 2020 20:00:50 -0700 Subject: [PATCH 4/5] integration test? --- .../StandbyTaskInEOSIntegrationTest.java | 25 +++++++++++++++++++ .../internals/StoreChangelogReaderTest.java | 3 +-- 2 files changed, 26 insertions(+), 2 deletions(-) create mode 100644 streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java new file mode 100644 index 0000000000000..4133a7ab98ae8 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java @@ -0,0 +1,25 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.integration; + +/** + * An integration test to verify the conversion of an active EOS + * task towards a standby task is safe across restarts of the application. + */ +public class StandbyTaskInEOSIntegrationTest { + +} diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index 116a6f2d5a023..ae540f7f8df2d 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -47,7 +47,6 @@ import java.util.Collection; import java.util.Collections; -import java.util.List; import java.util.Map; import java.util.Properties; import java.util.Set; @@ -947,7 +946,7 @@ public Map endOffsets(final Collection par @Test public void shouldNotThrowOnUnknownRevokedPartition() { - LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); + final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(); LogCaptureAppender.setClassLoggerToDebug(changelogReader.getClass()); try { From 3f4cfd7503c0959ed2453e88ee9f1c98d280ca71 Mon Sep 17 00:00:00 2001 From: abbccdda Date: Thu, 19 Mar 2020 20:45:25 -0700 Subject: [PATCH 5/5] add standby tasks integration --- .../processor/internals/StandbyTask.java | 1 + .../processor/internals/StreamTask.java | 2 +- .../StandbyTaskEOSIntegrationTest.java | 153 ++++++++++++++++++ .../StandbyTaskInEOSIntegrationTest.java | 25 --- .../internals/StoreChangelogReaderTest.java | 3 +- 5 files changed, 157 insertions(+), 27 deletions(-) create mode 100644 streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java delete mode 100644 streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java index 6e7c346f148e5..aecbe2fa67caf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StandbyTask.java @@ -210,6 +210,7 @@ private void close(final boolean clean) { offsetSnapshotSinceLastCommit = new HashMap<>(stateMgr.changelogOffsets()); } final boolean wipeStateStore = !clean && eosEnabled; + log.info("standby task clean {}, eos enabled {}", clean, eosEnabled); executeAndMaybeSwallow(clean, () -> StateManagerUtil.closeStateManager( diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java index e0062241d6023..a600858a7de5a 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java @@ -120,7 +120,7 @@ public StreamTask(final TaskId id, this.time = time; this.recordCollector = recordCollector; - eosEnabled = !StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); + eosEnabled = StreamsConfig.EXACTLY_ONCE.equals(config.getString(StreamsConfig.PROCESSING_GUARANTEE_CONFIG)); final String threadId = Thread.currentThread().getName(); closeTaskSensor = ThreadMetrics.closeTaskSensor(threadId, streamsMetrics); diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java new file mode 100644 index 0000000000000..bc22b5c7a5cc1 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java @@ -0,0 +1,153 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.streams.integration; + +import org.apache.kafka.clients.consumer.ConsumerConfig; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.serialization.IntegerSerializer; +import org.apache.kafka.common.serialization.Serdes; +import org.apache.kafka.common.utils.MockTime; +import org.apache.kafka.streams.KafkaStreams; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster; +import org.apache.kafka.streams.integration.utils.IntegrationTestUtils; +import org.apache.kafka.streams.kstream.Consumed; +import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.internals.StateDirectory; +import org.apache.kafka.streams.state.internals.OffsetCheckpoint; +import org.apache.kafka.test.TestUtils; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Test; + +import java.io.File; +import java.io.IOException; +import java.time.Duration; +import java.util.Collections; +import java.util.Properties; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.TimeUnit; + +import static org.apache.kafka.test.TestUtils.waitForCondition; +import static org.junit.Assert.assertTrue; + +/** + * An integration test to verify the conversion of a dirty-closed EOS + * task towards a standby task is safe across restarts of the application. + */ +public class StandbyTaskEOSIntegrationTest { + + private final String inputTopic = "input"; + + @ClassRule + public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(3); + + @Before + public void createTopics() throws Exception { + CLUSTER.createTopic(inputTopic, 1, 3); + } + + @Test + public void surviveWithOneTaskAsStandby() throws ExecutionException, InterruptedException, IOException { + IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp( + inputTopic, + Collections.singletonList( + new KeyValue<>(0, 0)), + TestUtils.producerConfig( + CLUSTER.bootstrapServers(), + IntegerSerializer.class, + IntegerSerializer.class, + new Properties()), + 10L); + + final String appId = "eos-test-app"; + final String stateDirPath = TestUtils.tempDirectory(appId).getPath(); + + final CountDownLatch instanceLatch = new CountDownLatch(1); + + final String stateDirPathOne = stateDirPath + "/" + appId + "-1/"; + final KafkaStreams streamInstanceOne = + buildStreamWithDirtyStateDir(appId, stateDirPathOne, instanceLatch); + + final String stateDirPathTwo = stateDirPath + "/" + appId + "-2/"; + final KafkaStreams streamInstanceTwo = + buildStreamWithDirtyStateDir(appId, stateDirPathTwo, instanceLatch); + + streamInstanceOne.start(); + + streamInstanceTwo.start(); + + // Wait for the record to be processed + assertTrue(instanceLatch.await(15, TimeUnit.SECONDS)); + + waitForCondition(() -> streamInstanceOne.state().equals(KafkaStreams.State.RUNNING), + "Stream instance one should be up and running by now"); + waitForCondition(() -> streamInstanceTwo.state().equals(KafkaStreams.State.RUNNING), + "Stream instance one should be up and running by now"); + + streamInstanceOne.close(Duration.ofSeconds(30)); + streamInstanceTwo.close(Duration.ofSeconds(30)); + + } + + private KafkaStreams buildStreamWithDirtyStateDir(final String appId, + final String stateDirPath, + final CountDownLatch recordProcessLatch) throws IOException { + + final StreamsBuilder builder = new StreamsBuilder(); + final TaskId taskId = new TaskId(0, 0); + + final Properties props = props(appId, stateDirPath); + + final StateDirectory stateDirectory = new StateDirectory( + new StreamsConfig(props), new MockTime(), true); + + new OffsetCheckpoint(new File(stateDirectory.directoryForTask(taskId), ".checkpoint")) + .write(Collections.singletonMap(new TopicPartition("unknown-topic", 0), 5L)); + + assertTrue(new File(stateDirectory.directoryForTask(taskId), + "rocksdb/KSTREAM-AGGREGATE-STATE-STORE-0000000001").mkdirs()); + + builder.stream(inputTopic, + Consumed.with(Serdes.Integer(), Serdes.Integer())) + .groupByKey() + .count() + .toStream() + .peek((key, value) -> recordProcessLatch.countDown()); + + return new KafkaStreams(builder.build(), props); + } + + private Properties props(final String appId, final String stateDirPath) { + final Properties streamsConfiguration = new Properties(); + streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, appId); + streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers()); + streamsConfiguration.put(StreamsConfig.CACHE_MAX_BYTES_BUFFERING_CONFIG, 0); + streamsConfiguration.put(StreamsConfig.STATE_DIR_CONFIG, stateDirPath); + streamsConfiguration.put(StreamsConfig.NUM_STANDBY_REPLICAS_CONFIG, 1); + streamsConfiguration.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE); + streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); + streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.Integer().getClass()); + streamsConfiguration.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, 1000); + streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + + return streamsConfiguration; + } +} diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java deleted file mode 100644 index 4133a7ab98ae8..0000000000000 --- a/streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskInEOSIntegrationTest.java +++ /dev/null @@ -1,25 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.kafka.streams.integration; - -/** - * An integration test to verify the conversion of an active EOS - * task towards a standby task is safe across restarts of the application. - */ -public class StandbyTaskInEOSIntegrationTest { - -} diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java index ae540f7f8df2d..076a1045da7b8 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StoreChangelogReaderTest.java @@ -950,7 +950,8 @@ public void shouldNotThrowOnUnknownRevokedPartition() { LogCaptureAppender.setClassLoggerToDebug(changelogReader.getClass()); try { - changelogReader.remove(Collections.singletonList(new TopicPartition("unknown", 0))); + changelogReader.remove( + Collections.singletonList(new TopicPartition("unknown", 0))); assertEquals(Collections.singletonList( "test-reader Changelog partition unknown-0 could not be found, " +