diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 2a93275083d..ca1c3d65763 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -194,7 +194,7 @@ + files="(KafkaStreams|KStreamImpl|KTableImpl|InternalTopologyBuilder|StreamsPartitionAssignor|StreamThread|IQv2StoreIntegrationTest|KStreamImplTest|RocksDBStore|StreamTask).java"/> diff --git a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java index b3723ef447d..3c6f4975cb8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -185,7 +185,6 @@ public class KafkaStreams implements AutoCloseable { protected final TopologyMetadata topologyMetadata; private final QueryableStoreProvider queryableStoreProvider; private final DelegatingStandbyUpdateListener delegatingStandbyUpdateListener; - private final LogContext logContext; GlobalStreamThread globalStreamThread; protected StateDirectory stateDirectory = null; @@ -643,9 +642,6 @@ public class KafkaStreams implements AutoCloseable { return; } - // all (alive) threads have received their assignment, close any remaining startup tasks, they're not needed - stateDirectory.closeStartupTasks(); - setState(State.RUNNING); } @@ -968,7 +964,7 @@ public class KafkaStreams implements AutoCloseable { } else { clientId = userClientId; } - logContext = new LogContext(String.format("stream-client [%s] ", clientId)); + final LogContext logContext = new LogContext(String.format("stream-client [%s] ", clientId)); this.log = logContext.logger(getClass()); topologyMetadata.setLog(logContext); @@ -1422,9 +1418,6 @@ public class KafkaStreams implements AutoCloseable { */ public synchronized void start() throws IllegalStateException, StreamsException { if (setState(State.REBALANCING)) { - log.debug("Initializing STANDBY tasks for existing local state"); - stateDirectory.initializeStartupTasks(topologyMetadata, streamsMetrics, logContext); - log.debug("Starting Streams client"); if (globalStreamThread != null) { diff --git a/streams/src/main/java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java b/streams/src/main/java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java index 65b7990dfe0..3f45a257b72 100644 --- a/streams/src/main/java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java +++ b/streams/src/main/java/org/apache/kafka/streams/internals/metrics/StreamsThreadMetricsDelegatingReporter.java @@ -57,7 +57,13 @@ public class StreamsThreadMetricsDelegatingReporter implements MetricsReporter { consumer.registerMetricForSubscription(metric); } } - + /* + The KafkaMetric object is a singleton shared by all StreamThread instances. + So we need to make sure we only pass metrics for the current StreamThread that contains this + MetricsReporter instance, which will register metrics with the embedded KafkaConsumer to pass + through the telemetry pipeline. + Otherwise, Kafka Streams would register multiple metrics for all StreamThreads. + */ private boolean tagMatchStreamOrStateUpdaterThreadId(final KafkaMetric metric) { final Map tags = metric.metricName().tags(); final boolean shouldInclude = tags.containsKey(THREAD_ID_TAG) && (tags.get(THREAD_ID_TAG).equals(threadId) || diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java index 3506845d288..7c5cc947c41 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorStateManager.java @@ -222,39 +222,6 @@ public class ProcessorStateManager implements StateManager { log.debug("Created state store manager for task {}", taskId); } - /** - * Special constructor used by {@link StateDirectory} to partially initialize startup tasks for local state, before - * they're assigned to a thread. When the task is assigned to a thread, the initialization of this StateManager is - * completed in {@link #assignToStreamThread(LogContext, ChangelogRegister, Collection)}. - */ - static ProcessorStateManager createStartupTaskStateManager(final TaskId taskId, - final boolean eosEnabled, - final LogContext logContext, - final StateDirectory stateDirectory, - final Map storeToChangelogTopic, - final Set sourcePartitions, - final boolean stateUpdaterEnabled) { - return new ProcessorStateManager(taskId, TaskType.STANDBY, eosEnabled, logContext, stateDirectory, null, storeToChangelogTopic, sourcePartitions, stateUpdaterEnabled); - } - - /** - * Standby tasks initialized for local state on-startup are only partially initialized, because they are not yet - * assigned to a StreamThread. Once assigned to a StreamThread, we complete their initialization here using the - * assigned StreamThread's context. - */ - void assignToStreamThread(final LogContext logContext, - final ChangelogRegister changelogReader, - final Collection sourcePartitions) { - if (this.changelogReader != null) { - throw new IllegalStateException("Attempted to replace an existing changelogReader on a StateManager without closing it."); - } - this.sourcePartitions.clear(); - this.log = logContext.logger(ProcessorStateManager.class); - this.logPrefix = logContext.logPrefix(); - this.changelogReader = changelogReader; - this.sourcePartitions.addAll(sourcePartitions); - } - void registerStateStores(final List allStores, final InternalProcessorContext processorContext) { processorContext.uninitialize(); for (final StateStore store : allStores) { @@ -347,7 +314,7 @@ public class ProcessorStateManager implements StateManager { } private void maybeRegisterStoreWithChangelogReader(final String storeName) { - if (isLoggingEnabled(storeName) && changelogReader != null) { + if (isLoggingEnabled(storeName)) { changelogReader.register(getStorePartition(storeName), this); } } @@ -616,7 +583,7 @@ public class ProcessorStateManager implements StateManager { public void close() throws ProcessorStateException { log.debug("Closing its state manager and all the registered state stores: {}", stores); - if (!stateUpdaterEnabled && changelogReader != null) { + if (!stateUpdaterEnabled) { changelogReader.unregister(getAllChangelogTopicPartitions()); } @@ -664,7 +631,7 @@ public class ProcessorStateManager implements StateManager { void recycle() { log.debug("Recycling state for {} task {}.", taskType, taskId); - if (!stateUpdaterEnabled && changelogReader != null) { + if (!stateUpdaterEnabled) { final List allChangelogs = getAllChangelogTopicPartitions(); changelogReader.unregister(allChangelogs); } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java index a95d20ddae0..070f732f4e0 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java @@ -16,18 +16,12 @@ */ package org.apache.kafka.streams.processor.internals; -import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.ProcessorStateException; import org.apache.kafka.streams.errors.StreamsException; -import org.apache.kafka.streams.errors.TaskCorruptedException; -import org.apache.kafka.streams.internals.StreamsConfigUtils; import org.apache.kafka.streams.processor.TaskId; -import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; -import org.apache.kafka.streams.state.internals.ThreadCache; import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import com.fasterxml.jackson.annotation.JsonProperty; @@ -49,18 +43,13 @@ import java.nio.file.attribute.PosixFilePermission; import java.nio.file.attribute.PosixFilePermissions; import java.util.ArrayList; import java.util.Arrays; -import java.util.Collection; import java.util.Collections; -import java.util.HashSet; +import java.util.HashMap; import java.util.List; -import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.UUID; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.Predicate; import java.util.regex.Pattern; import java.util.stream.Collectors; @@ -106,14 +95,11 @@ public class StateDirectory implements AutoCloseable { private final boolean hasPersistentStores; private final boolean hasNamedTopologies; - private final ConcurrentMap lockedTasksToOwner = new ConcurrentHashMap<>(); + private final HashMap lockedTasksToOwner = new HashMap<>(); private FileChannel stateDirLockChannel; private FileLock stateDirLock; - private final StreamsConfig config; - private final ConcurrentMap tasksForLocalState = new ConcurrentHashMap<>(); - /** * Ensures that the state base directory as well as the application's sub-directory are created. * @@ -132,7 +118,6 @@ public class StateDirectory implements AutoCloseable { this.hasPersistentStores = hasPersistentStores; this.hasNamedTopologies = hasNamedTopologies; this.appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); - this.config = config; final String stateDirName = config.getString(StreamsConfig.STATE_DIR_CONFIG); final File baseDir = new File(stateDirName); stateDir = new File(baseDir, appId); @@ -197,109 +182,6 @@ public class StateDirectory implements AutoCloseable { return stateDirLock != null; } - public void initializeStartupTasks(final TopologyMetadata topologyMetadata, - final StreamsMetricsImpl streamsMetrics, - final LogContext logContext) { - final List nonEmptyTaskDirectories = listNonEmptyTaskDirectories(); - if (hasPersistentStores && !nonEmptyTaskDirectories.isEmpty()) { - final ThreadCache dummyCache = new ThreadCache(logContext, 0, streamsMetrics); - final boolean eosEnabled = StreamsConfigUtils.eosEnabled(config); - final boolean stateUpdaterEnabled = StreamsConfig.InternalConfig.stateUpdaterEnabled(config.originals()); - - // discover all non-empty task directories in StateDirectory - for (final TaskDirectory taskDirectory : nonEmptyTaskDirectories) { - final String dirName = taskDirectory.file().getName(); - final TaskId id = parseTaskDirectoryName(dirName, taskDirectory.namedTopology()); - final ProcessorTopology subTopology = topologyMetadata.buildSubtopology(id); - - // we still check if the task's sub-topology is stateful, even though we know its directory contains state, - // because it's possible that the topology has changed since that data was written, and is now stateless - // this therefore prevents us from creating unnecessary Tasks just because of some left-over state - if (subTopology.hasStateWithChangelogs()) { - final Set inputPartitions = topologyMetadata.nodeToSourceTopics(id).values().stream() - .flatMap(Collection::stream) - .map(t -> new TopicPartition(t, id.partition())) - .collect(Collectors.toSet()); - final ProcessorStateManager stateManager = ProcessorStateManager.createStartupTaskStateManager( - id, - eosEnabled, - logContext, - this, - subTopology.storeToChangelogTopic(), - inputPartitions, - stateUpdaterEnabled - ); - - final InternalProcessorContext context = new ProcessorContextImpl( - id, - config, - stateManager, - streamsMetrics, - dummyCache - ); - - final Task task = new StandbyTask( - id, - inputPartitions, - subTopology, - topologyMetadata.taskConfig(id), - streamsMetrics, - stateManager, - this, - dummyCache, - context - ); - - try { - task.initializeIfNeeded(); - - tasksForLocalState.put(id, task); - } catch (final TaskCorruptedException e) { - // Task is corrupt - wipe it out (under EOS) and don't initialize a Standby for it - task.suspend(); - task.closeDirty(); - } - } - } - } - } - - public boolean hasStartupTasks() { - return !tasksForLocalState.isEmpty(); - } - - public Task removeStartupTask(final TaskId taskId) { - final Task task = tasksForLocalState.remove(taskId); - if (task != null) { - lockedTasksToOwner.replace(taskId, Thread.currentThread()); - } - return task; - } - - public void closeStartupTasks() { - closeStartupTasks(t -> true); - } - - private void closeStartupTasks(final Predicate predicate) { - if (!tasksForLocalState.isEmpty()) { - // "drain" Tasks first to ensure that we don't try to close Tasks that another thread is attempting to close - final Set drainedTasks = new HashSet<>(tasksForLocalState.size()); - for (final Map.Entry entry : tasksForLocalState.entrySet()) { - if (predicate.test(entry.getValue()) && removeStartupTask(entry.getKey()) != null) { - // only add to our list of drained Tasks if we exclusively "claimed" a Task from tasksForLocalState - // to ensure we don't accidentally try to drain the same Task multiple times from concurrent threads - drainedTasks.add(entry.getValue()); - } - } - - // now that we have exclusive ownership of the drained tasks, close them - for (final Task task : drainedTasks) { - task.suspend(); - task.closeClean(); - } - } - } - public UUID initializeProcessId() { if (!hasPersistentStores) { final UUID processId = UUID.randomUUID(); @@ -496,17 +378,9 @@ public class StateDirectory implements AutoCloseable { } } - /** - * Expose for tests. - */ - Thread lockOwner(final TaskId taskId) { - return lockedTasksToOwner.get(taskId); - } - @Override public void close() { if (hasPersistentStores) { - closeStartupTasks(); try { stateDirLock.release(); stateDirLockChannel.close(); @@ -624,7 +498,6 @@ public class StateDirectory implements AutoCloseable { ); if (namedTopologyDirs != null) { for (final File namedTopologyDir : namedTopologyDirs) { - closeStartupTasks(task -> task.id().topologyName().equals(parseNamedTopologyFromDirectory(namedTopologyDir.getName()))); final File[] contents = namedTopologyDir.listFiles(); if (contents != null && contents.length == 0) { try { @@ -662,7 +535,6 @@ public class StateDirectory implements AutoCloseable { log.debug("Tried to clear out the local state for NamedTopology {} but none was found", topologyName); } try { - closeStartupTasks(task -> task.id().topologyName().equals(topologyName)); Utils.delete(namedTopologyDir); } catch (final IOException e) { log.error("Hit an unexpected error while clearing local state for topology " + topologyName, e); 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 67d009b037f..3ec3bb78557 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,31 +333,6 @@ public class TaskManager { } } - private Map> assignStartupTasks(final Map> tasksToAssign, - final String threadLogPrefix, - final TopologyMetadata topologyMetadata, - final ChangelogRegister changelogReader) { - if (stateDirectory.hasStartupTasks()) { - final Map> assignedTasks = new HashMap<>(tasksToAssign.size()); - for (final Map.Entry> entry : tasksToAssign.entrySet()) { - final TaskId taskId = entry.getKey(); - final Task task = stateDirectory.removeStartupTask(taskId); - if (task != null) { - // replace our dummy values with the real ones, now we know our thread and assignment - final Set inputPartitions = entry.getValue(); - task.stateManager().assignToStreamThread(new LogContext(threadLogPrefix), changelogReader, inputPartitions); - updateInputPartitionsOfStandbyTaskIfTheyChanged(task, inputPartitions); - - assignedTasks.put(task, inputPartitions); - } - } - - return assignedTasks; - } else { - return Collections.emptyMap(); - } - } - /** * @throws TaskMigratedException if the task producer got fenced (EOS only) * @throws StreamsException fatal error while creating / initializing the task @@ -487,15 +462,6 @@ public class TaskManager { final Map> standbyTasksToCreate, final Map> tasksToRecycle, final Set tasksToCloseClean) { - final Map> startupStandbyTasksToRecycle = assignStartupTasks(activeTasksToCreate, logPrefix, topologyMetadata, changelogReader); - final Map> startupStandbyTasksToUse = assignStartupTasks(standbyTasksToCreate, logPrefix, topologyMetadata, changelogReader); - - // recycle the startup standbys to active - tasks.addStandbyTasks(startupStandbyTasksToRecycle.keySet()); - - // use startup Standbys as real Standby tasks - tasks.addStandbyTasks(startupStandbyTasksToUse.keySet()); - for (final Task task : tasks.allTasks()) { final TaskId taskId = task.id(); if (activeTasksToCreate.containsKey(taskId)) { @@ -550,7 +516,6 @@ public class TaskManager { final Set tasksToCloseClean, final Map failedTasks) { handleTasksPendingInitialization(); - handleStartupTaskReuse(activeTasksToCreate, standbyTasksToCreate, failedTasks); handleRestoringAndUpdatingTasks(activeTasksToCreate, standbyTasksToCreate, failedTasks); handleRunningAndSuspendedTasks(activeTasksToCreate, standbyTasksToCreate, tasksToRecycle, tasksToCloseClean); } @@ -568,34 +533,6 @@ public class TaskManager { } } - private void handleStartupTaskReuse(final Map> activeTasksToCreate, - final Map> standbyTasksToCreate, - final Map failedTasks) { - final Map> startupStandbyTasksToRecycle = assignStartupTasks(activeTasksToCreate, logPrefix, topologyMetadata, changelogReader); - final Map> startupStandbyTasksToUse = assignStartupTasks(standbyTasksToCreate, logPrefix, topologyMetadata, changelogReader); - - // recycle the startup standbys to active, and remove them from the set of actives that need to be created - if (!startupStandbyTasksToRecycle.isEmpty()) { - final Set tasksToCloseDirty = new HashSet<>(); - for (final Map.Entry> entry : startupStandbyTasksToRecycle.entrySet()) { - final Task task = entry.getKey(); - recycleTaskFromStateUpdater(task, entry.getValue(), tasksToCloseDirty, failedTasks); - activeTasksToCreate.remove(task.id()); - } - - // if any standby tasks failed to recycle, close them dirty - tasksToCloseDirty.forEach(task -> - closeTaskDirty(task, false) - ); - } - - // use startup Standbys as real Standby tasks - if (!startupStandbyTasksToUse.isEmpty()) { - tasks.addPendingTasksToInit(startupStandbyTasksToUse.keySet()); - startupStandbyTasksToUse.keySet().forEach(task -> standbyTasksToCreate.remove(task.id())); - } - } - private void handleRunningAndSuspendedTasks(final Map> activeTasksToCreate, final Map> standbyTasksToCreate, final Map> tasksToRecycle, diff --git a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java index d417b5be2ed..e1212b7f308 100644 --- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java @@ -400,44 +400,6 @@ public class KafkaStreamsTest { } } - @Test - public void shouldInitializeTasksForLocalStateOnStart() { - prepareStreams(); - prepareStreamThread(streamThreadOne, 1); - prepareStreamThread(streamThreadTwo, 2); - - try (final MockedConstruction constructed = mockConstruction(StateDirectory.class, - (mock, context) -> when(mock.initializeProcessId()).thenReturn(UUID.randomUUID()))) { - try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) { - assertEquals(1, constructed.constructed().size()); - final StateDirectory stateDirectory = constructed.constructed().get(0); - verify(stateDirectory, times(0)).initializeStartupTasks(any(), any(), any()); - streams.start(); - verify(stateDirectory, times(1)).initializeStartupTasks(any(), any(), any()); - } - } - } - - @Test - public void shouldCloseStartupTasksAfterFirstRebalance() throws Exception { - prepareStreams(); - final AtomicReference state1 = prepareStreamThread(streamThreadOne, 1); - final AtomicReference state2 = prepareStreamThread(streamThreadTwo, 2); - prepareThreadState(streamThreadOne, state1); - prepareThreadState(streamThreadTwo, state2); - try (final MockedConstruction constructed = mockConstruction(StateDirectory.class, - (mock, context) -> when(mock.initializeProcessId()).thenReturn(UUID.randomUUID()))) { - try (final KafkaStreams streams = new KafkaStreams(getBuilderWithSource().build(), props, supplier, time)) { - assertEquals(1, constructed.constructed().size()); - final StateDirectory stateDirectory = constructed.constructed().get(0); - streams.setStateListener(streamsStateListener); - streams.start(); - waitForCondition(() -> streams.state() == State.RUNNING, "Streams never started."); - verify(stateDirectory, times(1)).closeStartupTasks(); - } - } - } - @Test public void stateShouldTransitToRunningIfNonDeadThreadsBackToRunning() throws Exception { prepareStreams(); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java index f850093d389..515bff1df16 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StateDirectoryTest.java @@ -17,20 +17,14 @@ package org.apache.kafka.streams.processor.internals; import org.apache.kafka.common.TopicPartition; -import org.apache.kafka.common.metrics.Metrics; import org.apache.kafka.common.utils.LogCaptureAppender; -import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.common.utils.Utils; import org.apache.kafka.streams.StreamsConfig; -import org.apache.kafka.streams.TopologyConfig; import org.apache.kafka.streams.errors.ProcessorStateException; -import org.apache.kafka.streams.processor.StateStore; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.StateDirectory.TaskDirectory; -import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; -import org.apache.kafka.test.MockKeyValueStore; import org.apache.kafka.test.TestUtils; import com.fasterxml.jackson.annotation.JsonProperty; @@ -39,8 +33,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import org.mockito.ArgumentMatchers; -import org.mockito.Mockito; import java.io.BufferedWriter; import java.io.File; @@ -79,7 +71,6 @@ import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.endsWith; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.CoreMatchers.hasItem; -import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.CoreMatchers.not; import static org.hamcrest.MatcherAssert.assertThat; @@ -96,7 +87,6 @@ public class StateDirectoryTest { private final MockTime time = new MockTime(); private File stateDir; private final String applicationId = "applicationId"; - private StreamsConfig config; private StateDirectory directory; private File appDir; @@ -105,14 +95,15 @@ public class StateDirectoryTest { if (!createStateDirectory) { cleanup(); } - config = new StreamsConfig(new Properties() { - { - put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); - put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234"); - put(StreamsConfig.STATE_DIR_CONFIG, stateDir.getPath()); - } - }); - directory = new StateDirectory(config, time, createStateDirectory, hasNamedTopology); + directory = new StateDirectory( + new StreamsConfig(new Properties() { + { + put(StreamsConfig.APPLICATION_ID_CONFIG, applicationId); + put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "dummy:1234"); + put(StreamsConfig.STATE_DIR_CONFIG, stateDir.getPath()); + } + }), + time, createStateDirectory, hasNamedTopology); appDir = new File(stateDir, applicationId); } @@ -821,144 +812,6 @@ public class StateDirectoryTest { assertThat(directory.initializeProcessId(), equalTo(processId)); } - @Test - public void shouldNotInitializeStandbyTasksWhenNoLocalState() { - final TaskId taskId = new TaskId(0, 0); - initializeStartupTasks(new TaskId(0, 0), false); - assertFalse(directory.hasStartupTasks()); - assertNull(directory.removeStartupTask(taskId)); - assertFalse(directory.hasStartupTasks()); - } - - @Test - public void shouldInitializeStandbyTasksForLocalState() { - final TaskId taskId = new TaskId(0, 0); - initializeStartupTasks(new TaskId(0, 0), true); - assertTrue(directory.hasStartupTasks()); - assertNotNull(directory.removeStartupTask(taskId)); - assertFalse(directory.hasStartupTasks()); - assertNull(directory.removeStartupTask(taskId)); - } - - @Test - public void shouldNotAssignStartupTasksWeDontHave() { - final TaskId taskId = new TaskId(0, 0); - initializeStartupTasks(taskId, false); - final Task task = directory.removeStartupTask(taskId); - assertNull(task); - } - - private class FakeStreamThread extends Thread { - private final TaskId taskId; - private final AtomicReference result; - - private FakeStreamThread(final TaskId taskId, final AtomicReference result) { - this.taskId = taskId; - this.result = result; - } - - @Override - public void run() { - result.set(directory.removeStartupTask(taskId)); - } - } - - @Test - public void shouldAssignStartupTaskToStreamThread() throws InterruptedException { - final TaskId taskId = new TaskId(0, 0); - - initializeStartupTasks(taskId, true); - - // main thread owns the newly initialized tasks - assertThat(directory.lockOwner(taskId), is(Thread.currentThread())); - - // spawn off a "fake" StreamThread, so we can verify the lock was updated to the correct thread - final AtomicReference result = new AtomicReference<>(); - final Thread streamThread = new FakeStreamThread(taskId, result); - streamThread.start(); - streamThread.join(); - final Task task = result.get(); - - assertNotNull(task); - assertThat(task, instanceOf(StandbyTask.class)); - - // verify the owner of the task directory lock has been shifted over to our assigned StreamThread - assertThat(directory.lockOwner(taskId), is(instanceOf(FakeStreamThread.class))); - } - - @Test - public void shouldUnlockStartupTasksOnClose() { - final TaskId taskId = new TaskId(0, 0); - initializeStartupTasks(taskId, true); - - assertEquals(Thread.currentThread(), directory.lockOwner(taskId)); - directory.closeStartupTasks(); - assertNull(directory.lockOwner(taskId)); - } - - @Test - public void shouldCloseStartupTasksOnDirectoryClose() { - final StateStore store = initializeStartupTasks(new TaskId(0, 0), true); - - assertTrue(directory.hasStartupTasks()); - assertTrue(store.isOpen()); - - directory.close(); - - assertFalse(directory.hasStartupTasks()); - assertFalse(store.isOpen()); - } - - @Test - public void shouldNotCloseStartupTasksOnAutoCleanUp() { - // we need to set this because the auto-cleanup uses the last-modified time from the filesystem, - // which can't be mocked - time.setCurrentTimeMs(System.currentTimeMillis()); - - final StateStore store = initializeStartupTasks(new TaskId(0, 0), true); - - assertTrue(directory.hasStartupTasks()); - assertTrue(store.isOpen()); - - time.sleep(10000); - - directory.cleanRemovedTasks(1000); - - assertTrue(directory.hasStartupTasks()); - assertTrue(store.isOpen()); - } - - private StateStore initializeStartupTasks(final TaskId taskId, final boolean createTaskDir) { - directory.initializeProcessId(); - final TopologyMetadata metadata = Mockito.mock(TopologyMetadata.class); - final TopologyConfig topologyConfig = new TopologyConfig(config); - - final StateStore store = new MockKeyValueStore("test", true); - - if (createTaskDir) { - final File taskDir = directory.getOrCreateDirectoryForTask(taskId); - final File storeDir = new File(taskDir, store.name()); - storeDir.mkdir(); - } - - final ProcessorTopology processorTopology = new ProcessorTopology( - Collections.emptyList(), - Collections.emptyMap(), - Collections.emptyMap(), - Collections.singletonList(store), - Collections.emptyList(), - Collections.singletonMap(store.name(), store.name() + "-changelog"), - Collections.emptySet(), - Collections.emptyMap() - ); - Mockito.when(metadata.buildSubtopology(ArgumentMatchers.any())).thenReturn(processorTopology); - Mockito.when(metadata.taskConfig(ArgumentMatchers.any())).thenReturn(topologyConfig.getTaskConfig()); - - directory.initializeStartupTasks(metadata, new StreamsMetricsImpl(new Metrics(), "test", "processId", time), new LogContext("test")); - - return store; - } - private static class FutureStateDirectoryProcessFile { @JsonProperty diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java index 1b4143eceee..9eed49110bb 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java @@ -1163,7 +1163,6 @@ public class StreamThreadTest { new StreamsMetricsImpl(metrics, CLIENT_ID, PROCESS_ID.toString(), mockTime); final TopologyMetadata topologyMetadata = new TopologyMetadata(internalTopologyBuilder, config); topologyMetadata.buildAndRewriteTopology(); - stateDirectory = new StateDirectory(config, mockTime, true, false); final TaskManager taskManager = new TaskManager( new MockTime(), @@ -1175,7 +1174,7 @@ public class StreamThreadTest { new Tasks(new LogContext()), topologyMetadata, null, - stateDirectory, + null, stateUpdater, schedulingTaskManager ) { diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java index 26a1523131b..00b4d5d8a07 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/TaskManagerTest.java @@ -61,8 +61,6 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; import org.junit.jupiter.api.io.TempDir; -import org.junit.jupiter.params.ParameterizedTest; -import org.junit.jupiter.params.provider.ValueSource; import org.mockito.InOrder; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; @@ -4719,138 +4717,6 @@ public class TaskManagerTest { assertEquals(taskManager.notPausedTasks().size(), 0); } - @Test - public void shouldRecycleStartupTasksFromStateDirectoryAsActive() { - final StandbyTask startupTask = standbyTask(taskId00, taskId00ChangelogPartitions).build(); - final StreamTask activeTask = statefulTask(taskId00, taskId00ChangelogPartitions).build(); - when(activeTaskCreator.createActiveTaskFromStandby(eq(startupTask), eq(taskId00Partitions), any())) - .thenReturn(activeTask); - - when(stateDirectory.hasStartupTasks()).thenReturn(true, false); - when(stateDirectory.removeStartupTask(taskId00)).thenReturn(startupTask, (Task) null); - - taskManager.handleAssignment(taskId00Assignment, Collections.emptyMap()); - - // ensure we recycled our existing startup Standby into an Active task - verify(activeTaskCreator).createActiveTaskFromStandby(eq(startupTask), eq(taskId00Partitions), any()); - - // ensure we didn't construct any new Tasks - verify(activeTaskCreator).createTasks(any(), eq(Collections.emptyMap())); - verify(standbyTaskCreator).createTasks(Collections.emptyMap()); - verifyNoMoreInteractions(activeTaskCreator); - verifyNoMoreInteractions(standbyTaskCreator); - - // verify the recycled task is now being used as an assigned Active - assertEquals(Collections.singletonMap(taskId00, activeTask), taskManager.activeTaskMap()); - assertEquals(Collections.emptyMap(), taskManager.standbyTaskMap()); - } - - @Test - public void shouldUseStartupTasksFromStateDirectoryAsStandby() { - final StandbyTask startupTask = standbyTask(taskId00, taskId00ChangelogPartitions).build(); - - when(stateDirectory.hasStartupTasks()).thenReturn(true, true, false); - when(stateDirectory.removeStartupTask(taskId00)).thenReturn(startupTask, (Task) null); - - taskManager.handleAssignment(Collections.emptyMap(), taskId00Assignment); - - // ensure we used our existing startup Task directly as a Standby - verify(startupTask).resume(); - - // ensure we didn't construct any new Tasks, or recycle an existing Task; we only used the one we already have - verify(activeTaskCreator).createTasks(any(), eq(Collections.emptyMap())); - verify(standbyTaskCreator).createTasks(Collections.emptyMap()); - verifyNoMoreInteractions(activeTaskCreator); - verifyNoMoreInteractions(standbyTaskCreator); - - // verify the startup Standby is now being used as an assigned Standby - assertEquals(Collections.emptyMap(), taskManager.activeTaskMap()); - assertEquals(Collections.singletonMap(taskId00, startupTask), taskManager.standbyTaskMap()); - } - - @Test - public void shouldRecycleStartupTasksFromStateDirectoryAsActiveWithStateUpdater() { - final Tasks taskRegistry = new Tasks(new LogContext()); - final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, taskRegistry, true); - final StandbyTask startupTask = standbyTask(taskId00, taskId00ChangelogPartitions).build(); - - final StreamTask activeTask = statefulTask(taskId00, taskId00ChangelogPartitions).build(); - when(activeTaskCreator.createActiveTaskFromStandby(eq(startupTask), eq(taskId00Partitions), any())) - .thenReturn(activeTask); - - when(stateDirectory.hasStartupTasks()).thenReturn(true, false); - when(stateDirectory.removeStartupTask(taskId00)).thenReturn(startupTask, (Task) null); - - taskManager.handleAssignment(taskId00Assignment, Collections.emptyMap()); - - // ensure we used our existing startup Task directly as a Standby - assertTrue(taskRegistry.hasPendingTasksToInit()); - assertEquals(Collections.singleton(activeTask), taskRegistry.drainPendingTasksToInit()); - - // we're using a mock StateUpdater here, so now that we've drained the task from the queue of startup tasks to init - // let's "add" it to our mock StateUpdater - when(stateUpdater.tasks()).thenReturn(Collections.singleton(activeTask)); - when(stateUpdater.standbyTasks()).thenReturn(Collections.emptySet()); - - // ensure we recycled our existing startup Standby into an Active task - verify(activeTaskCreator).createActiveTaskFromStandby(eq(startupTask), eq(taskId00Partitions), any()); - - // ensure we didn't construct any new Tasks - verify(activeTaskCreator).createTasks(any(), eq(Collections.emptyMap())); - verify(standbyTaskCreator).createTasks(Collections.emptyMap()); - verifyNoMoreInteractions(activeTaskCreator); - verifyNoMoreInteractions(standbyTaskCreator); - - // verify the recycled task is now being used as an assigned Active - assertEquals(Collections.singletonMap(taskId00, activeTask), taskManager.activeTaskMap()); - assertEquals(Collections.emptyMap(), taskManager.standbyTaskMap()); - } - - @Test - public void shouldUseStartupTasksFromStateDirectoryAsStandbyWithStateUpdater() { - final Tasks taskRegistry = new Tasks(new LogContext()); - final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, taskRegistry, true); - final StandbyTask startupTask = standbyTask(taskId00, taskId00ChangelogPartitions).build(); - - when(stateDirectory.hasStartupTasks()).thenReturn(true, true, false); - when(stateDirectory.removeStartupTask(taskId00)).thenReturn(startupTask, (Task) null); - - assertFalse(taskRegistry.hasPendingTasksToInit()); - - taskManager.handleAssignment(Collections.emptyMap(), taskId00Assignment); - - // ensure we used our existing startup Task directly as a Standby - assertTrue(taskRegistry.hasPendingTasksToInit()); - assertEquals(Collections.singleton(startupTask), taskRegistry.drainPendingTasksToInit()); - - // we're using a mock StateUpdater here, so now that we've drained the task from the queue of startup tasks to init - // let's "add" it to our mock StateUpdater - when(stateUpdater.tasks()).thenReturn(Collections.singleton(startupTask)); - when(stateUpdater.standbyTasks()).thenReturn(Collections.singleton(startupTask)); - - // ensure we didn't construct any new Tasks, or recycle an existing Task; we only used the one we already have - verify(activeTaskCreator).createTasks(any(), eq(Collections.emptyMap())); - verify(standbyTaskCreator).createTasks(Collections.emptyMap()); - verifyNoMoreInteractions(activeTaskCreator); - verifyNoMoreInteractions(standbyTaskCreator); - - // verify the startup Standby is now being used as an assigned Standby - assertEquals(Collections.emptyMap(), taskManager.activeTaskMap()); - assertEquals(Collections.singletonMap(taskId00, startupTask), taskManager.standbyTaskMap()); - } - - @ParameterizedTest - @ValueSource(booleans = {true, false}) - public void shouldStartStateUpdaterOnInit(final boolean stateUpdaterEnabled) { - final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, stateUpdaterEnabled); - taskManager.init(); - if (stateUpdaterEnabled) { - verify(stateUpdater).start(); - } else { - verify(stateUpdater, never()).start(); - } - } - private static KafkaFutureImpl completedFuture() { final KafkaFutureImpl futureDeletedRecords = new KafkaFutureImpl<>(); futureDeletedRecords.complete(null);