diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index 21cd5f4cc5ebc..ff5414f67d3e5 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -195,7 +195,7 @@ + files="(StreamsPartitionAssignorTest|StreamThreadTest|StreamTaskTest|TaskManagerTest|TopologyTestDriverTest).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 8446c35240964..bf12927bda45c 100644 --- a/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java +++ b/streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java @@ -827,7 +827,7 @@ private KafkaStreams(final InternalTopologyBuilder internalTopologyBuilder, (hasGlobalTopology && globalTaskTopology.hasPersistentGlobalStore()); try { - stateDirectory = new StateDirectory(config, time, hasPersistentStores); + stateDirectory = new StateDirectory(config, time, hasPersistentStores, internalTopologyBuilder.hasNamedTopologies()); processId = stateDirectory.initializeProcessId(); } catch (final ProcessorStateException fatal) { throw new StreamsException(fatal); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java index 838cff93a8d51..db7c91cdfe05f 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java @@ -2039,6 +2039,11 @@ public synchronized List fullSourceTopicNames() { return maybeDecorateInternalSourceTopics(sourceTopicNames); } + public boolean hasNamedTopologies() { + // TODO KAFKA-12648: covered by Pt. 2 + return false; + } + // following functions are for test only public synchronized Set sourceTopicNames() { return sourceTopicNames; 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 f3a24aa0b52da..5977503409500 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,13 +16,20 @@ */ package org.apache.kafka.streams.processor.internals; - 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.processor.TaskId; + +import java.io.FileFilter; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.stream.Collectors; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -56,6 +63,7 @@ public class StateDirectory { private static final Pattern TASK_DIR_PATH_NAME = Pattern.compile("\\d+_\\d+"); + private static final Pattern NAMED_TOPOLOGY_DIR_PATH_NAME = Pattern.compile("__.+__"); // named topology dirs follow '__Topology-Name__' private static final Logger log = LoggerFactory.getLogger(StateDirectory.class); static final String LOCK_FILE_NAME = ".lock"; @@ -84,6 +92,7 @@ public StateDirectoryProcessFile() { private final String appId; private final File stateDir; private final boolean hasPersistentStores; + private final boolean hasNamedTopologies; private final HashMap lockedTasksToOwner = new HashMap<>(); @@ -98,13 +107,15 @@ public StateDirectoryProcessFile() { * @param hasPersistentStores only when the application's topology does have stores persisted on local file * system, we would go ahead and auto-create the corresponding application / task / store * directories whenever necessary; otherwise no directories would be created. + * @param hasNamedTopologies whether this application is composed of independent named topologies * * @throws ProcessorStateException if the base state directory or application state directory does not exist * and could not be created when hasPersistentStores is enabled. */ - public StateDirectory(final StreamsConfig config, final Time time, final boolean hasPersistentStores) { + public StateDirectory(final StreamsConfig config, final Time time, final boolean hasPersistentStores, final boolean hasNamedTopologies) { this.time = time; this.hasPersistentStores = hasPersistentStores; + this.hasNamedTopologies = hasNamedTopologies; this.appId = config.getString(StreamsConfig.APPLICATION_ID_CONFIG); final String stateDirName = config.getString(StreamsConfig.STATE_DIR_CONFIG); final File baseDir = new File(stateDirName); @@ -132,6 +143,12 @@ public StateDirectory(final StreamsConfig config, final Time time, final boolean } } + public StateDirectory(final StreamsConfig config, final Time time, final boolean hasPersistentStores) { + // TODO KAFKA-12648: Explicitly set hasNamedTopology in each test and remove this constructor + // Will be done in Pt. 2 as we want some of these integration tests to use named topologies + this(config, time, hasPersistentStores, false); + } + private void configurePermissions(final File file) { final Path path = file.toPath(); if (path.getFileSystem().supportedFileAttributeViews().contains("posix")) { @@ -208,16 +225,22 @@ public UUID initializeProcessId() { /** * Get or create the directory for the provided {@link TaskId}. * @return directory for the {@link TaskId} - * @throws ProcessorStateException if the task directory does not exists and could not be created + * @throws ProcessorStateException if the task directory does not exist and could not be created */ public File getOrCreateDirectoryForTask(final TaskId taskId) { - final File taskDir = new File(stateDir, taskId.toString()); + final File taskParentDir = getTaskDirectoryParentName(taskId); + final File taskDir = new File(taskParentDir, StateManagerUtil.toTaskDirString(taskId)); if (hasPersistentStores && !taskDir.exists()) { synchronized (taskDirCreationLock) { // to avoid a race condition, we need to check again if the directory does not exist: // otherwise, two threads might pass the outer `if` (and enter the `then` block), // one blocks on `synchronized` while the other creates the directory, // and the blocking one fails when trying to create it after it's unblocked + if (!taskParentDir.exists() && !taskParentDir.mkdir()) { + throw new ProcessorStateException( + String.format("Parent [%s] of task directory [%s] doesn't exist and couldn't be created", + taskParentDir.getPath(), taskDir.getPath())); + } if (!taskDir.exists() && !taskDir.mkdir()) { throw new ProcessorStateException( String.format("task directory [%s] doesn't exist and couldn't be created", taskDir.getPath())); @@ -227,6 +250,18 @@ public File getOrCreateDirectoryForTask(final TaskId taskId) { return taskDir; } + private File getTaskDirectoryParentName(final TaskId taskId) { + final String namedTopology = taskId.namedTopology(); + if (namedTopology != null) { + if (!hasNamedTopologies) { + throw new IllegalStateException("Tried to lookup taskId with named topology, but StateDirectory thinks hasNamedTopologies = false"); + } + return new File(stateDir, "__" + namedTopology + "__"); + } else { + return stateDir; + } + } + /** * @return The File handle for the checkpoint in the given task's directory */ @@ -387,18 +422,18 @@ public synchronized void cleanRemovedTasks(final long cleanupDelayMs) { } private void cleanRemovedTasksCalledByCleanerThread(final long cleanupDelayMs) { - for (final File taskDir : listNonEmptyTaskDirectories()) { - final String dirName = taskDir.getName(); - final TaskId id = parseTaskDirectoryName(dirName, null); + for (final TaskDirectory taskDir : listAllTaskDirectories()) { + final String dirName = taskDir.file().getName(); + final TaskId id = parseTaskDirectoryName(dirName, taskDir.namedTopology()); if (!lockedTasksToOwner.containsKey(id)) { try { if (lock(id)) { final long now = time.milliseconds(); - final long lastModifiedMs = taskDir.lastModified(); + final long lastModifiedMs = taskDir.file().lastModified(); if (now > lastModifiedMs + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as {}ms has elapsed (cleanup delay is {}ms).", logPrefix(), dirName, id, now - lastModifiedMs, cleanupDelayMs); - Utils.delete(taskDir); + Utils.delete(taskDir.file()); } } } catch (final IOException exception) { @@ -412,35 +447,83 @@ private void cleanRemovedTasksCalledByCleanerThread(final long cleanupDelayMs) { } } } + // Ok to ignore returned exception as it should be swallowed + maybeCleanEmptyNamedTopologyDirs(true); + } + + /** + * Cleans up any leftover named topology directories that are empty, if any exist + * @param logExceptionAsWarn if true, an exception will be logged as a warning + * if false, an exception will be logged as error + * @return the first IOException to be encountered + */ + private IOException maybeCleanEmptyNamedTopologyDirs(final boolean logExceptionAsWarn) { + if (!hasNamedTopologies) { + return null; + } + + final AtomicReference firstException = new AtomicReference<>(null); + final File[] namedTopologyDirs = stateDir.listFiles(pathname -> + pathname.isDirectory() && NAMED_TOPOLOGY_DIR_PATH_NAME.matcher(pathname.getName()).matches() + ); + if (namedTopologyDirs != null) { + for (final File namedTopologyDir : namedTopologyDirs) { + final File[] contents = namedTopologyDir.listFiles(); + if (contents != null && contents.length == 0) { + try { + Utils.delete(namedTopologyDir); + } catch (final IOException exception) { + if (logExceptionAsWarn) { + log.warn( + String.format("%sSwallowed the following exception during deletion of named topology directory %s", + logPrefix(), namedTopologyDir.getName()), + exception + ); + } else { + log.error( + String.format("%s Failed to delete named topology directory %s with exception:", + logPrefix(), namedTopologyDir.getName()), + exception + ); + } + firstException.compareAndSet(null, exception); + } + } + } + } + return firstException.get(); } private void cleanStateAndTaskDirectoriesCalledByUser() throws Exception { if (!lockedTasksToOwner.isEmpty()) { - log.warn("Found some still-locked task directories when user requested to cleaning up the state, " - + "since Streams is not running any more these will be ignored to complete the cleanup"); + log.warn("Found some still-locked task directories when user requested to cleaning up the state, " + + "since Streams is not running any more these will be ignored to complete the cleanup"); } final AtomicReference firstException = new AtomicReference<>(); - for (final File taskDir : listAllTaskDirectories()) { - final String dirName = taskDir.getName(); - final TaskId id = parseTaskDirectoryName(dirName, null); + for (final TaskDirectory taskDir : listAllTaskDirectories()) { + final String dirName = taskDir.file().getName(); + final TaskId id = parseTaskDirectoryName(dirName, taskDir.namedTopology()); try { - log.info("{} Deleting state directory {} for task {} as user calling cleanup.", - logPrefix(), dirName, id); - + log.info("{} Deleting task directory {} for {} as user calling cleanup.", + logPrefix(), dirName, id); + if (lockedTasksToOwner.containsKey(id)) { log.warn("{} Task {} in state directory {} was still locked by {}", - logPrefix(), dirName, id, lockedTasksToOwner.get(id)); + logPrefix(), dirName, id, lockedTasksToOwner.get(id)); } - Utils.delete(taskDir); + Utils.delete(taskDir.file()); } catch (final IOException exception) { log.error( - String.format("%s Failed to delete state directory %s for task %s with exception:", - logPrefix(), dirName, id), + String.format("%s Failed to delete task directory %s for %s with exception:", + logPrefix(), dirName, id), exception ); firstException.compareAndSet(null, exception); } } + + firstException.compareAndSet(null, maybeCleanEmptyNamedTopologyDirs(false)); + final Exception exception = firstException.get(); if (exception != null) { throw exception; @@ -451,39 +534,56 @@ private void cleanStateAndTaskDirectoriesCalledByUser() throws Exception { * List all of the task directories that are non-empty * @return The list of all the non-empty local directories for stream tasks */ - File[] listNonEmptyTaskDirectories() { - final File[] taskDirectories; - if (!hasPersistentStores || !stateDir.exists()) { - taskDirectories = new File[0]; - } else { - taskDirectories = - stateDir.listFiles(pathname -> { - if (!pathname.isDirectory() || !TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches()) { - return false; - } else { - return !taskDirIsEmpty(pathname); - } - }); - } - - return taskDirectories == null ? new File[0] : taskDirectories; + List listNonEmptyTaskDirectories() { + return listTaskDirectories(pathname -> { + if (!pathname.isDirectory() || !TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches()) { + return false; + } else { + return !taskDirIsEmpty(pathname); + } + }); } /** - * List all of the task directories + * List all of the task directories along with their parent directory if they belong to a named topology * @return The list of all the existing local directories for stream tasks */ - File[] listAllTaskDirectories() { - final File[] taskDirectories; - if (!hasPersistentStores || !stateDir.exists()) { - taskDirectories = new File[0]; - } else { - taskDirectories = - stateDir.listFiles(pathname -> pathname.isDirectory() - && TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches()); + List listAllTaskDirectories() { + return listTaskDirectories(pathname -> pathname.isDirectory() && TASK_DIR_PATH_NAME.matcher(pathname.getName()).matches()); + } + + private List listTaskDirectories(final FileFilter filter) { + final List taskDirectories = new ArrayList<>(); + if (hasPersistentStores && stateDir.exists()) { + if (hasNamedTopologies) { + for (final File namedTopologyDir : listNamedTopologyDirs()) { + final String namedTopology = parseNamedTopologyFromDirectory(namedTopologyDir.getName()); + final File[] taskDirs = namedTopologyDir.listFiles(filter); + if (taskDirs != null) { + taskDirectories.addAll(Arrays.stream(taskDirs) + .map(f -> new TaskDirectory(f, namedTopology)).collect(Collectors.toList())); + } + } + } else { + final File[] taskDirs = + stateDir.listFiles(filter); + if (taskDirs != null) { + taskDirectories.addAll(Arrays.stream(taskDirs) + .map(f -> new TaskDirectory(f, null)).collect(Collectors.toList())); + } + } } - return taskDirectories == null ? new File[0] : taskDirectories; + return taskDirectories; + } + + private List listNamedTopologyDirs() { + final File[] namedTopologyDirectories = stateDir.listFiles(f -> f.getName().startsWith("__") && f.getName().endsWith("__")); + return namedTopologyDirectories != null ? Arrays.asList(namedTopologyDirectories) : Collections.emptyList(); + } + + private String parseNamedTopologyFromDirectory(final String dirName) { + return dirName.substring(2, dirName.length() - 2); } private FileLock tryLock(final FileChannel channel) throws IOException { @@ -494,4 +594,40 @@ private FileLock tryLock(final FileChannel channel) throws IOException { } } + public static class TaskDirectory { + private final File file; + private final String namedTopology; // may be null if hasNamedTopologies = false + + TaskDirectory(final File file, final String namedTopology) { + this.file = file; + this.namedTopology = namedTopology; + } + + public File file() { + return file; + } + + public String namedTopology() { + return namedTopology; + } + + @Override + public boolean equals(final Object o) { + if (this == o) { + return true; + } + if (o == null || getClass() != o.getClass()) { + return false; + } + final TaskDirectory that = (TaskDirectory) o; + return file.equals(that.file) && + Objects.equals(namedTopology, that.namedTopology); + } + + @Override + public int hashCode() { + return Objects.hash(file, namedTopology); + } + } + } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java index 16d73838b9104..83451422e0c59 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java @@ -1255,6 +1255,7 @@ public void onAssignment(final Assignment assignment, final ConsumerGroupMetadat case 7: case 8: case 9: + case 10: validateActiveTaskEncoding(partitions, info, logPrefix); activeTasks = getActiveTasks(partitions, info); 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 5d5217e4b1c45..7f32526683eaf 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 @@ -36,9 +36,11 @@ import org.apache.kafka.streams.errors.TaskIdFormatException; import org.apache.kafka.streams.errors.TaskMigratedException; import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.internals.StateDirectory.TaskDirectory; import org.apache.kafka.streams.processor.internals.Task.State; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; + import org.slf4j.Logger; import java.io.File; @@ -682,9 +684,11 @@ private void tryToLockAllNonEmptyTaskDirectories() { // current set of actually-locked tasks. lockedTaskDirectories.clear(); - for (final File dir : stateDirectory.listNonEmptyTaskDirectories()) { + for (final TaskDirectory taskDir : stateDirectory.listNonEmptyTaskDirectories()) { + final File dir = taskDir.file(); + final String namedTopology = taskDir.namedTopology(); try { - final TaskId id = parseTaskDirectoryName(dir.getName(), null); + final TaskId id = parseTaskDirectoryName(dir.getName(), namedTopology); if (stateDirectory.lock(id)) { lockedTaskDirectories.add(id); if (!tasks.owned(id)) { diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java index 6ecc4ada3827c..ecf0b2541fcaf 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfo.java @@ -184,6 +184,7 @@ public ByteBuffer encode() { case 7: case 8: case 9: + case 10: out.writeInt(usedVersion); out.writeInt(commonlySupportedVersion); encodeActiveAndStandbyTaskAssignment(out); @@ -359,6 +360,7 @@ public static AssignmentInfo decode(final ByteBuffer data) { case 7: case 8: case 9: + case 10: commonlySupportedVersion = in.readInt(); assignmentInfo = new AssignmentInfo(usedVersion, commonlySupportedVersion); decodeActiveTasks(assignmentInfo, in); diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java index f54aca7115455..467dbfa9248f2 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ConsumerProtocolUtils.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.streams.processor.internals.assignment; +import org.apache.kafka.streams.errors.TaskAssignmentException; import org.apache.kafka.streams.processor.TaskId; import java.io.DataInputStream; @@ -43,6 +44,8 @@ public static void writeTaskIdTo(final TaskId taskId, final DataOutputStream out } else { out.writeInt(0); } + } else if (taskId.namedTopology() != null) { + throw new TaskAssignmentException("Named topologies are not compatible with protocol version " + version); } } @@ -78,6 +81,8 @@ public static void writeTaskIdTo(final TaskId taskId, final ByteBuffer buf, fina } else { buf.putInt(0); } + } else if (taskId.namedTopology() != null) { + throw new TaskAssignmentException("Named topologies are not compatible with protocol version " + version); } } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/StreamsAssignmentProtocolVersions.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/StreamsAssignmentProtocolVersions.java index 0273999ed4074..42bb766f97398 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/StreamsAssignmentProtocolVersions.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/StreamsAssignmentProtocolVersions.java @@ -20,10 +20,18 @@ public final class StreamsAssignmentProtocolVersions { public static final int UNKNOWN = -1; public static final int EARLIEST_PROBEABLE_VERSION = 3; public static final int MIN_NAMED_TOPOLOGY_VERSION = 10; - public static final int LATEST_SUPPORTED_VERSION = 9; - // When changing the versions: - // 1) Update the version_probing_message and end_of_upgrade_message in streams_upgrade_test.py::StreamsUpgradeTest.test_version_probing_upgrade - // 2) Add a unit test in SubscriptionInfoTest and/or AssignmentInfoTest + public static final int LATEST_SUPPORTED_VERSION = 10; + /* + * Any time you modify the subscription or assignment info, you need to bump the latest supported version, unless + * the version has already been bumped within the current release cycle. + * + * Last version bump: May 2021, before 3.0 + * + * When changing the version: + * 1) Update the version_probing_message and end_of_upgrade_message in streams_upgrade_test.py::StreamsUpgradeTest.test_version_probing_upgrade + * 2) Add a unit test in SubscriptionInfoTest and/or AssignmentInfoTest + * 3) Note the date & corresponding Kafka version of this bump + */ private StreamsAssignmentProtocolVersions() {} } diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java index 971439caa014d..cf9df3c29a356 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfo.java @@ -31,6 +31,7 @@ import org.apache.kafka.streams.internals.generated.SubscriptionInfoData.TaskOffsetSum; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.Task; + import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +43,7 @@ import java.util.stream.Collectors; import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION; +import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.MIN_NAMED_TOPOLOGY_VERSION; public class SubscriptionInfo { private static final Logger LOG = LoggerFactory.getLogger(SubscriptionInfo.class); @@ -94,8 +96,8 @@ public SubscriptionInfo(final int version, if (version >= 2) { data.setUserEndPoint(userEndPoint == null - ? new byte[0] - : userEndPoint.getBytes(StandardCharsets.UTF_8)); + ? new byte[0] + : userEndPoint.getBytes(StandardCharsets.UTF_8)); } if (version >= 3) { data.setLatestSupportedVersion(latestSupportedVersion); @@ -109,7 +111,9 @@ public SubscriptionInfo(final int version, this.data = data; - if (version >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) { + if (version >= MIN_NAMED_TOPOLOGY_VERSION) { + setTaskOffsetSumDataWithNamedTopologiesFromTaskOffsetSumMap(taskOffsetSums); + } else if (version >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) { setTaskOffsetSumDataFromTaskOffsetSumMap(taskOffsetSums); } else { setPrevAndStandbySetsFromParsedTaskOffsetSumMap(taskOffsetSums); @@ -125,10 +129,27 @@ public int errorCode() { return data.errorCode(); } + // For version > MIN_NAMED_TOPOLOGY_VERSION + private void setTaskOffsetSumDataWithNamedTopologiesFromTaskOffsetSumMap(final Map taskOffsetSums) { + data.setTaskOffsetSums(taskOffsetSums.entrySet().stream().map(t -> { + final SubscriptionInfoData.TaskOffsetSum taskOffsetSum = new SubscriptionInfoData.TaskOffsetSum(); + final TaskId task = t.getKey(); + taskOffsetSum.setTopicGroupId(task.subtopology()); + taskOffsetSum.setPartition(task.partition()); + taskOffsetSum.setNamedTopology(task.namedTopology()); + taskOffsetSum.setOffsetSum(t.getValue()); + return taskOffsetSum; + }).collect(Collectors.toList())); + } + + // For MIN_NAMED_TOPOLOGY_VERSION > version > MIN_VERSION_OFFSET_SUM_SUBSCRIPTION private void setTaskOffsetSumDataFromTaskOffsetSumMap(final Map taskOffsetSums) { final Map> topicGroupIdToPartitionOffsetSum = new HashMap<>(); for (final Map.Entry taskEntry : taskOffsetSums.entrySet()) { final TaskId task = taskEntry.getKey(); + if (task.namedTopology() != null) { + throw new TaskAssignmentException("Named topologies are not compatible with older protocol versions"); + } topicGroupIdToPartitionOffsetSum.computeIfAbsent(task.subtopology(), t -> new ArrayList<>()).add( new SubscriptionInfoData.PartitionToOffsetSum() .setPartition(task.partition()) @@ -143,11 +164,15 @@ private void setTaskOffsetSumDataFromTaskOffsetSumMap(final Map ta }).collect(Collectors.toList())); } + // For MIN_VERSION_OFFSET_SUM_SUBSCRIPTION > version private void setPrevAndStandbySetsFromParsedTaskOffsetSumMap(final Map taskOffsetSums) { final Set prevTasks = new HashSet<>(); final Set standbyTasks = new HashSet<>(); for (final Map.Entry taskOffsetSum : taskOffsetSums.entrySet()) { + if (taskOffsetSum.getKey().namedTopology() != null) { + throw new TaskAssignmentException("Named topologies are not compatible with older protocol versions"); + } if (taskOffsetSum.getValue() == Task.LATEST_OFFSET) { prevTasks.add(taskOffsetSum.getKey()); } else { @@ -218,12 +243,20 @@ public Map taskOffsetSums() { taskOffsetSumsCache = new HashMap<>(); if (data.version() >= MIN_VERSION_OFFSET_SUM_SUBSCRIPTION) { for (final TaskOffsetSum taskOffsetSum : data.taskOffsetSums()) { - for (final PartitionToOffsetSum partitionOffsetSum : taskOffsetSum.partitionToOffsetSum()) { + if (data.version() >= MIN_NAMED_TOPOLOGY_VERSION) { taskOffsetSumsCache.put( new TaskId(taskOffsetSum.topicGroupId(), - partitionOffsetSum.partition()), - partitionOffsetSum.offsetSum() - ); + taskOffsetSum.partition(), + taskOffsetSum.namedTopology()), + taskOffsetSum.offsetSum()); + } else { + for (final PartitionToOffsetSum partitionOffsetSum : taskOffsetSum.partitionToOffsetSum()) { + taskOffsetSumsCache.put( + new TaskId(taskOffsetSum.topicGroupId(), + partitionOffsetSum.partition()), + partitionOffsetSum.offsetSum() + ); + } } } } else { diff --git a/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopology.java b/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopology.java index 18e89f77d11d9..c1588a930bdf8 100644 --- a/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopology.java +++ b/streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/NamedTopology.java @@ -26,9 +26,7 @@ public class NamedTopology extends Topology { private final Logger log = LoggerFactory.getLogger(NamedTopology.class); private String name; - - - + void setTopologyName(final String newTopologyName) { if (name != null) { log.error("Unable to set topologyName = {} because the name is already set to {}", newTopologyName, name); @@ -41,7 +39,6 @@ public String name() { return name; } - public List sourceTopics() { return super.internalTopologyBuilder.fullSourceTopicNames(); } diff --git a/streams/src/main/resources/common/message/SubscriptionInfoData.json b/streams/src/main/resources/common/message/SubscriptionInfoData.json index 13a62bd503878..15868638c6e0c 100644 --- a/streams/src/main/resources/common/message/SubscriptionInfoData.json +++ b/streams/src/main/resources/common/message/SubscriptionInfoData.json @@ -15,7 +15,7 @@ { "name": "SubscriptionInfoData", - "validVersions": "1-9", + "validVersions": "1-10", "fields": [ { "name": "version", @@ -33,6 +33,7 @@ "versions": "1+", "type": "uuid" }, + /***** Protocol version 1-6 only (after 6 this is encoded in task offset sum map) *****/ { "name": "prevTasks", "versions": "1-6", @@ -43,6 +44,7 @@ "versions": "1-6", "type": "[]TaskId" }, + /***************/ { "name": "userEndPoint", "versions": "2+", @@ -65,18 +67,19 @@ } ], "commonStructs": [ + // TaskId was only used from 1-6, after 6 we encode each field of the TaskId separately along with the other information for that map entry { "name": "TaskId", - "versions": "1+", + "versions": "1-6", "fields": [ { "name": "topicGroupId", - "versions": "1+", + "versions": "1-6", "type": "int32" }, { "name": "partition", - "versions": "1+", + "versions": "1-6", "type": "int32" } ] @@ -90,25 +93,45 @@ "versions": "7+", "type": "int32" }, + // Prior to version 10, in 7-9, the below fields (partition and offsetSum) were encoded via the nested + // partitionToOffsetSum struct. In 10+ all fields are encoded directly in the TaskOffsetSum struct + { + "name": "partition", + "versions": "10+", + "type": "int32" + }, + { + "name": "offsetSum", + "versions": "10+", + "type": "int64" + }, + { + "name": "namedTopology", + "versions": "10+", + "nullableVersions": "10+", + "ignorable": "false", // namedTopology is not ignorable because if you do, a TaskId may not be unique + "type": "string" + }, { "name": "partitionToOffsetSum", - "versions": "7+", + "versions": "7-9", "type": "[]PartitionToOffsetSum" } ] }, + { "name": "PartitionToOffsetSum", - "versions": "7+", + "versions": "7-9", "fields": [ { "name": "partition", - "versions": "7+", + "versions": "7-9", "type": "int32" }, { "name": "offsetSum", - "versions": "7+", + "versions": "7-9", "type": "int64" } ] 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 492a145d33ba2..22e4bfafa5047 100644 --- a/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java @@ -97,6 +97,8 @@ import static org.apache.kafka.streams.state.QueryableStoreTypes.keyValueStore; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.safeUniqueTestName; import static org.apache.kafka.streams.integration.utils.IntegrationTestUtils.waitForApplicationState; + +import static org.easymock.EasyMock.anyBoolean; import static org.apache.kafka.test.TestUtils.waitForCondition; import static org.easymock.EasyMock.anyInt; import static org.easymock.EasyMock.anyLong; @@ -948,7 +950,8 @@ public void shouldCleanupOldStateDirs() throws Exception { PowerMock.expectNew(StateDirectory.class, anyObject(StreamsConfig.class), anyObject(Time.class), - EasyMock.eq(true) + EasyMock.eq(true), + anyBoolean() ).andReturn(stateDirectory); EasyMock.expect(stateDirectory.initializeProcessId()).andReturn(UUID.randomUUID()); stateDirectory.close(); @@ -1119,7 +1122,8 @@ private void startStreamsAndCheckDirExists(final Topology topology, PowerMock.expectNew(StateDirectory.class, anyObject(StreamsConfig.class), anyObject(Time.class), - EasyMock.eq(shouldFilesExist) + EasyMock.eq(shouldFilesExist), + anyBoolean() ).andReturn(stateDirectory); EasyMock.expect(stateDirectory.initializeProcessId()).andReturn(UUID.randomUUID()); diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java new file mode 100644 index 0000000000000..5ea3d391f9580 --- /dev/null +++ b/streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java @@ -0,0 +1,32 @@ +/* + * 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; + +public class NamedTopologyIntegrationTest { + //TODO KAFKA-12648 + /** + * Things to test in Pt. 2 - Introduce TopologyMetadata to wrap InternalTopologyBuilders of named topologies: + * 1. Verify changelog & repartition topics decorated with named topology + * 2. Make sure app run and works with + * -multiple subtopologies + * -persistent state + * -multi-partition input & output topics + * -standbys + * -piped input and verified output records + * 3. Is the task assignment balanced? Does KIP-441/warmup replica placement work as intended? + */ +} 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 fb1a486b573ed..7b68b8a731fad 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 @@ -22,6 +22,8 @@ import java.io.FileOutputStream; import java.io.OutputStreamWriter; import java.nio.charset.StandardCharsets; +import java.util.HashSet; +import java.util.List; import java.util.UUID; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.utils.MockTime; @@ -29,19 +31,19 @@ import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.errors.ProcessorStateException; import org.apache.kafka.streams.processor.TaskId; +import org.apache.kafka.streams.processor.internals.StateDirectory.TaskDirectory; import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; import org.apache.kafka.test.TestUtils; + import org.junit.After; import org.junit.Before; import org.junit.Test; import java.io.File; import java.io.IOException; -import java.nio.channels.FileChannel; import java.nio.file.Files; import java.nio.file.Path; -import java.nio.file.StandardOpenOption; import java.nio.file.attribute.PosixFilePermission; import java.time.Duration; import java.util.Arrays; @@ -59,9 +61,14 @@ 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.mkSet; -import static org.apache.kafka.streams.processor.internals.StateDirectory.LOCK_FILE_NAME; import static org.apache.kafka.streams.processor.internals.StateDirectory.PROCESS_FILE_NAME; import static org.apache.kafka.streams.processor.internals.StateManagerUtil.CHECKPOINT_FILE_NAME; +import static org.apache.kafka.streams.processor.internals.StateManagerUtil.toTaskDirString; + +import static java.util.Collections.emptyList; +import static java.util.Collections.emptySet; +import static java.util.Collections.singleton; +import static java.util.Collections.singletonList; import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.endsWith; import static org.hamcrest.CoreMatchers.equalTo; @@ -85,7 +92,7 @@ public class StateDirectoryTest { private StateDirectory directory; private File appDir; - private void initializeStateDirectory(final boolean createStateDirectory) throws IOException { + private void initializeStateDirectory(final boolean createStateDirectory, final boolean hasNamedTopology) throws IOException { stateDir = new File(TestUtils.IO_TMP_DIR, "kafka-" + TestUtils.randomString(5)); if (!createStateDirectory) { cleanup(); @@ -98,13 +105,13 @@ private void initializeStateDirectory(final boolean createStateDirectory) throws put(StreamsConfig.STATE_DIR_CONFIG, stateDir.getPath()); } }), - time, createStateDirectory); + time, createStateDirectory, hasNamedTopology); appDir = new File(stateDir, applicationId); } @Before public void before() throws IOException { - initializeStateDirectory(true); + initializeStateDirectory(true, false); } @After @@ -248,24 +255,6 @@ public void shouldLockMultipleTaskDirectories() { directory.unlock(taskId2); } - @Test - public void shouldReleaseTaskStateDirectoryLock() throws IOException { - final TaskId taskId = new TaskId(0, 0); - final File taskDirectory = directory.getOrCreateDirectoryForTask(taskId); - - directory.lock(taskId); - directory.unlock(taskId); - - try ( - final FileChannel channel = FileChannel.open( - new File(taskDirectory, LOCK_FILE_NAME).toPath(), - StandardOpenOption.CREATE, - StandardOpenOption.WRITE) - ) { - channel.tryLock(); - } - } - @Test public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() { final TaskId task0 = new TaskId(0, 0); @@ -279,28 +268,24 @@ public void shouldCleanUpTaskStateDirectoriesThatAreNotCurrentlyLocked() { directory.lock(task0); directory.lock(task1); - final File dir0 = new File(appDir, task0.toString()); - final File dir1 = new File(appDir, task1.toString()); - final File dir2 = new File(appDir, task2.toString()); + final TaskDirectory dir0 = new TaskDirectory(new File(appDir, toTaskDirString(task0)), null); + final TaskDirectory dir1 = new TaskDirectory(new File(appDir, toTaskDirString(task1)), null); + final TaskDirectory dir2 = new TaskDirectory(new File(appDir, toTaskDirString(task2)), null); - Set files = Arrays.stream( - Objects.requireNonNull(directory.listAllTaskDirectories())).collect(Collectors.toSet()); - assertEquals(mkSet(dir0, dir1, dir2), files); + List files = directory.listAllTaskDirectories(); + assertEquals(mkSet(dir0, dir1, dir2), new HashSet<>(files)); - files = Arrays.stream( - Objects.requireNonNull(directory.listNonEmptyTaskDirectories())).collect(Collectors.toSet()); - assertEquals(mkSet(dir0, dir1, dir2), files); + files = directory.listNonEmptyTaskDirectories(); + assertEquals(mkSet(dir0, dir1, dir2), new HashSet<>(files)); time.sleep(5000); directory.cleanRemovedTasks(0); - files = Arrays.stream( - Objects.requireNonNull(directory.listAllTaskDirectories())).collect(Collectors.toSet()); - assertEquals(mkSet(dir0, dir1), files); + files = directory.listAllTaskDirectories(); + assertEquals(mkSet(dir0, dir1), new HashSet<>(files)); - files = Arrays.stream( - Objects.requireNonNull(directory.listNonEmptyTaskDirectories())).collect(Collectors.toSet()); - assertEquals(mkSet(dir0, dir1), files); + files = directory.listNonEmptyTaskDirectories(); + assertEquals(mkSet(dir0, dir1), new HashSet<>(files)); } finally { directory.unlock(task0); directory.unlock(task1); @@ -315,29 +300,29 @@ public void shouldCleanupStateDirectoriesWhenLastModifiedIsLessThanNowMinusClean final int cleanupDelayMs = 60000; directory.cleanRemovedTasks(cleanupDelayMs); assertTrue(dir.exists()); - assertEquals(1, directory.listAllTaskDirectories().length); - assertEquals(1, directory.listNonEmptyTaskDirectories().length); + assertEquals(1, directory.listAllTaskDirectories().size()); + assertEquals(1, directory.listNonEmptyTaskDirectories().size()); time.sleep(cleanupDelayMs + 1000); directory.cleanRemovedTasks(cleanupDelayMs); assertFalse(dir.exists()); - assertEquals(0, directory.listAllTaskDirectories().length); - assertEquals(0, directory.listNonEmptyTaskDirectories().length); + assertEquals(0, directory.listAllTaskDirectories().size()); + assertEquals(0, directory.listNonEmptyTaskDirectories().size()); } @Test public void shouldCleanupObsoleteTaskDirectoriesAndDeleteTheDirectoryItself() { final File dir = directory.getOrCreateDirectoryForTask(new TaskId(2, 0)); assertTrue(new File(dir, "store").mkdir()); - assertEquals(1, directory.listAllTaskDirectories().length); - assertEquals(1, directory.listNonEmptyTaskDirectories().length); + assertEquals(1, directory.listAllTaskDirectories().size()); + assertEquals(1, directory.listNonEmptyTaskDirectories().size()); try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StateDirectory.class)) { time.sleep(5000); directory.cleanRemovedTasks(0); assertFalse(dir.exists()); - assertEquals(0, directory.listAllTaskDirectories().length); - assertEquals(0, directory.listNonEmptyTaskDirectories().length); + assertEquals(0, directory.listAllTaskDirectories().size()); + assertEquals(0, directory.listNonEmptyTaskDirectories().size()); assertThat( appender.getMessages(), hasItem(containsString("Deleting obsolete state directory")) @@ -354,15 +339,15 @@ public void shouldNotRemoveNonTaskDirectoriesAndFiles() { @Test public void shouldReturnEmptyArrayForNonPersistentApp() throws IOException { - initializeStateDirectory(false); - assertTrue(Arrays.asList(directory.listAllTaskDirectories()).isEmpty()); + initializeStateDirectory(false, false); + assertTrue(directory.listAllTaskDirectories().isEmpty()); } @Test public void shouldReturnEmptyArrayIfStateDirDoesntExist() throws IOException { cleanup(); assertFalse(stateDir.exists()); - assertTrue(Arrays.asList(directory.listAllTaskDirectories()).isEmpty()); + assertTrue(directory.listAllTaskDirectories().isEmpty()); } @Test @@ -384,29 +369,25 @@ public void shouldReturnEmptyArrayIfListFilesReturnsNull() throws IOException { assertTrue(appDir.createNewFile()); assertTrue(appDir.exists()); assertNull(appDir.listFiles()); - assertEquals(0, directory.listAllTaskDirectories().length); + assertEquals(0, directory.listAllTaskDirectories().size()); } @Test public void shouldOnlyListNonEmptyTaskDirectories() throws IOException { TestUtils.tempDirectory(stateDir.toPath(), "foo"); - final File taskDir1 = directory.getOrCreateDirectoryForTask(new TaskId(0, 0)); - final File taskDir2 = directory.getOrCreateDirectoryForTask(new TaskId(0, 1)); + final TaskDirectory taskDir1 = new TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 0)), null); + final TaskDirectory taskDir2 = new TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 1)), null); - final File storeDir = new File(taskDir1, "store"); + final File storeDir = new File(taskDir1.file(), "store"); assertTrue(storeDir.mkdir()); - assertEquals(mkSet(taskDir1, taskDir2), Arrays.stream( - directory.listAllTaskDirectories()).collect(Collectors.toSet())); - assertEquals(mkSet(taskDir1), Arrays.stream( - directory.listNonEmptyTaskDirectories()).collect(Collectors.toSet())); + assertThat(mkSet(taskDir1, taskDir2), equalTo(new HashSet<>(directory.listAllTaskDirectories()))); + assertThat(singletonList(taskDir1), equalTo(directory.listNonEmptyTaskDirectories())); - Utils.delete(taskDir1); + Utils.delete(taskDir1.file()); - assertEquals(mkSet(taskDir2), Arrays.stream( - directory.listAllTaskDirectories()).collect(Collectors.toSet())); - assertEquals(Collections.emptySet(), Arrays.stream( - directory.listNonEmptyTaskDirectories()).collect(Collectors.toSet())); + assertThat(singleton(taskDir2), equalTo(new HashSet<>(directory.listAllTaskDirectories()))); + assertThat(emptyList(), equalTo(directory.listNonEmptyTaskDirectories())); } @Test @@ -479,14 +460,14 @@ public void shouldCleanupAllTaskDirectoriesIncludingGlobalOne() { directory.clean(); - assertEquals(Collections.emptySet(), Arrays.stream( + assertEquals(emptySet(), Arrays.stream( Objects.requireNonNull(appDir.listFiles())).collect(Collectors.toSet())); } @Test public void shouldNotCreateBaseDirectory() throws IOException { try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StateDirectory.class)) { - initializeStateDirectory(false); + initializeStateDirectory(false, false); assertThat(stateDir.exists(), is(false)); assertThat(appDir.exists(), is(false)); assertThat(appender.getMessages(), @@ -496,7 +477,7 @@ public void shouldNotCreateBaseDirectory() throws IOException { @Test public void shouldNotCreateTaskStateDirectory() throws IOException { - initializeStateDirectory(false); + initializeStateDirectory(false, false); final TaskId taskId = new TaskId(0, 0); final File taskDirectory = directory.getOrCreateDirectoryForTask(taskId); assertFalse(taskDirectory.exists()); @@ -504,14 +485,14 @@ public void shouldNotCreateTaskStateDirectory() throws IOException { @Test public void shouldNotCreateGlobalStateDirectory() throws IOException { - initializeStateDirectory(false); + initializeStateDirectory(false, false); final File globalStateDir = directory.globalStateDir(); assertFalse(globalStateDir.exists()); } @Test public void shouldLockTaskStateDirectoryWhenDirectoryCreationDisabled() throws IOException { - initializeStateDirectory(false); + initializeStateDirectory(false, false); final TaskId taskId = new TaskId(0, 0); assertTrue(directory.lock(taskId)); } @@ -582,7 +563,8 @@ public void shouldLogTempDirMessage() { ) ), new MockTime(), - true + true, + false ); assertThat( appender.getMessages(), @@ -593,6 +575,111 @@ public void shouldLogTempDirMessage() { } } + /************* Named Topology Tests *************/ + + @Test + public void shouldCreateTaskDirectoriesUnderNamedTopologyDirs() throws IOException { + initializeStateDirectory(true, true); + + directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology1")); + directory.getOrCreateDirectoryForTask(new TaskId(0, 1, "topology1")); + directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology2")); + + assertThat(new File(appDir, "__topology1__").exists(), is(true)); + assertThat(new File(appDir, "__topology1__").isDirectory(), is(true)); + assertThat(new File(appDir, "__topology2__").exists(), is(true)); + assertThat(new File(appDir, "__topology2__").isDirectory(), is(true)); + + assertThat(new File(new File(appDir, "__topology1__"), "0_0").exists(), is(true)); + assertThat(new File(new File(appDir, "__topology1__"), "0_0").isDirectory(), is(true)); + assertThat(new File(new File(appDir, "__topology1__"), "0_1").exists(), is(true)); + assertThat(new File(new File(appDir, "__topology1__"), "0_1").isDirectory(), is(true)); + assertThat(new File(new File(appDir, "__topology2__"), "0_0").exists(), is(true)); + assertThat(new File(new File(appDir, "__topology2__"), "0_0").isDirectory(), is(true)); + } + + @Test + public void shouldOnlyListNonEmptyTaskDirectoriesInNamedTopologies() throws IOException { + initializeStateDirectory(true, true); + + TestUtils.tempDirectory(appDir.toPath(), "foo"); + final TaskDirectory taskDir1 = new TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology1")), "topology1"); + final TaskDirectory taskDir2 = new TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 1, "topology1")), "topology1"); + final TaskDirectory taskDir3 = new TaskDirectory(directory.getOrCreateDirectoryForTask(new TaskId(0, 0, "topology2")), "topology2"); + + final File storeDir = new File(taskDir1.file(), "store"); + assertTrue(storeDir.mkdir()); + + assertThat(new HashSet<>(directory.listAllTaskDirectories()), equalTo(mkSet(taskDir1, taskDir2, taskDir3))); + assertThat(directory.listNonEmptyTaskDirectories(), equalTo(singletonList(taskDir1))); + + Utils.delete(taskDir1.file()); + + assertThat(new HashSet<>(directory.listAllTaskDirectories()), equalTo(mkSet(taskDir2, taskDir3))); + assertThat(directory.listNonEmptyTaskDirectories(), equalTo(emptyList())); + } + + @Test + public void shouldRemoveNonEmptyNamedTopologyDirsWhenCallingClean() throws Exception { + initializeStateDirectory(true, true); + final File taskDir = directory.getOrCreateDirectoryForTask(new TaskId(2, 0, "topology1")); + final File namedTopologyDir = new File(appDir, "__topology1__"); + + assertThat(taskDir.exists(), is(true)); + assertThat(namedTopologyDir.exists(), is(true)); + directory.clean(); + assertThat(taskDir.exists(), is(false)); + assertThat(namedTopologyDir.exists(), is(false)); + } + + @Test + public void shouldRemoveEmptyNamedTopologyDirsWhenCallingClean() throws IOException { + initializeStateDirectory(true, true); + final File namedTopologyDir = new File(appDir, "__topology1__"); + assertThat(namedTopologyDir.mkdir(), is(true)); + assertThat(namedTopologyDir.exists(), is(true)); + directory.clean(); + assertThat(namedTopologyDir.exists(), is(false)); + } + + @Test + public void shouldNotRemoveDirsThatDoNotMatchNamedTopologyDirsWhenCallingClean() throws IOException { + initializeStateDirectory(true, true); + final File someDir = new File(appDir, "_not-a-valid-named-topology_dir_name_"); + assertThat(someDir.mkdir(), is(true)); + assertThat(someDir.exists(), is(true)); + directory.clean(); + assertThat(someDir.exists(), is(true)); + } + + @Test + public void shouldCleanupObsoleteTaskDirectoriesInNamedTopologiesAndDeleteTheParentDirectories() throws IOException { + initializeStateDirectory(true, true); + + final File taskDir = directory.getOrCreateDirectoryForTask(new TaskId(2, 0, "topology1")); + final File namedTopologyDir = new File(appDir, "__topology1__"); + assertThat(namedTopologyDir.exists(), is(true)); + assertThat(taskDir.exists(), is(true)); + assertTrue(new File(taskDir, "store").mkdir()); + assertThat(directory.listAllTaskDirectories().size(), is(1)); + assertThat(directory.listNonEmptyTaskDirectories().size(), is(1)); + + try (final LogCaptureAppender appender = LogCaptureAppender.createAndRegister(StateDirectory.class)) { + time.sleep(5000); + directory.cleanRemovedTasks(0); + assertThat(taskDir.exists(), is(false)); + assertThat(namedTopologyDir.exists(), is(false)); + assertThat(directory.listAllTaskDirectories().size(), is(0)); + assertThat(directory.listNonEmptyTaskDirectories().size(), is(0)); + assertThat( + appender.getMessages(), + hasItem(containsString("Deleting obsolete state directory")) + ); + } + } + + /************************************************/ + @Test public void shouldPersistProcessIdAcrossRestart() { final UUID processId = directory.initializeProcessId(); 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 7c27f74ab44dd..60dd1907fa1bf 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 @@ -43,11 +43,14 @@ import org.apache.kafka.streams.errors.TaskMigratedException; 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.StreamThread.ProcessingMode; import org.apache.kafka.streams.processor.internals.Task.State; import org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl; import org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender; import org.apache.kafka.streams.state.internals.OffsetCheckpoint; + +import java.util.ArrayList; import org.easymock.EasyMock; import org.easymock.EasyMockRunner; import org.easymock.Mock; @@ -210,7 +213,7 @@ public void shouldIdempotentlyUpdateSubscriptionFromActiveAssignment() { @Test public void shouldNotLockAnythingIfStateDirIsEmpty() { - expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(new File[0]).once(); + expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(new ArrayList<>()).once(); replay(stateDirectory); taskManager.handleRebalanceStart(singleton("topic")); @@ -823,7 +826,7 @@ public Map prepareCommit() { @Test public void shouldNotAttemptToCommitInHandleCorruptedDuringARebalance() { final ProcessorStateManager stateManager = EasyMock.createNiceMock(ProcessorStateManager.class); - expect(stateDirectory.listNonEmptyTaskDirectories()).andStubReturn(new File[0]); + expect(stateDirectory.listNonEmptyTaskDirectories()).andStubReturn(new ArrayList<>()); final StateMachineTask corruptedActive = new StateMachineTask(taskId00, taskId00Partitions, true, stateManager); @@ -2032,7 +2035,7 @@ public void shouldHandleRebalanceEvents() { expect(consumer.assignment()).andReturn(assignment); consumer.pause(assignment); expectLastCall(); - expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(new File[0]); + expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(new ArrayList<>()); replay(consumer, stateDirectory); assertThat(taskManager.isRebalanceInProgress(), is(false)); taskManager.handleRebalanceStart(emptySet()); @@ -3244,9 +3247,9 @@ private static KafkaFutureImpl completedFuture() { } private void makeTaskFolders(final String... names) throws Exception { - final File[] taskFolders = new File[names.length]; + final ArrayList taskFolders = new ArrayList<>(names.length); for (int i = 0; i < names.length; ++i) { - taskFolders[i] = testFolder.newFolder(names[i]); + taskFolders.add(new TaskDirectory(testFolder.newFolder(names[i]), null)); } expect(stateDirectory.listNonEmptyTaskDirectories()).andReturn(taskFolders).once(); } diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfoTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfoTest.java index a652759c271bd..a4c9534d29aef 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfoTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentInfoTest.java @@ -17,6 +17,7 @@ package org.apache.kafka.streams.processor.internals.assignment; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.streams.errors.TaskAssignmentException; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.state.HostInfo; import org.junit.Test; @@ -30,11 +31,19 @@ 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.mkSet; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_0_1; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_1_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T1_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T1_0_1; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T2_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T2_2_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_0_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_0_1; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_1_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_1_1; import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION; +import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.MIN_NAMED_TOPOLOGY_VERSION; import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.UNKNOWN; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertThrows; @@ -44,13 +53,30 @@ public class AssignmentInfoTest { TASK_0_0, TASK_0_1, TASK_1_0, - TASK_1_0); + TASK_1_0 + ); private final Map> standbyTasks = mkMap( mkEntry(TASK_1_0, mkSet(new TopicPartition("t1", 0), new TopicPartition("t2", 0))), mkEntry(TASK_1_1, mkSet(new TopicPartition("t1", 1), new TopicPartition("t2", 1))) ); + private static final List NAMED_ACTIVE_TASKS = Arrays.asList( + NAMED_TASK_T0_0_1, + NAMED_TASK_T0_1_0, + NAMED_TASK_T0_1_0, + NAMED_TASK_T1_0_1, + NAMED_TASK_T1_0_1, + NAMED_TASK_T2_0_0, + NAMED_TASK_T2_2_0 + ); + + private static final Map> NAMED_STANDBY_TASKS = mkMap( + mkEntry(NAMED_TASK_T0_0_0, mkSet(new TopicPartition("t0-1", 0), new TopicPartition("t0-2", 0))), + mkEntry(NAMED_TASK_T0_0_1, mkSet(new TopicPartition("t0-1", 1), new TopicPartition("t0-2", 1))), + mkEntry(NAMED_TASK_T1_0_0, mkSet(new TopicPartition("t1-1", 0), new TopicPartition("t1-2", 0))) + ); + private final Map> activeAssignment = mkMap( mkEntry(new HostInfo("localhost", 8088), mkSet(new TopicPartition("t0", 0), @@ -144,6 +170,47 @@ public void shouldEncodeAndDecodeVersion7() { assertEquals(expectedInfo, AssignmentInfo.decode(info.encode())); } + @Test + public void shouldEncodeAndDecodeVersion8() { + final AssignmentInfo info = + new AssignmentInfo(8, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + final AssignmentInfo expectedInfo = + new AssignmentInfo(8, LATEST_SUPPORTED_VERSION, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + assertEquals(expectedInfo, AssignmentInfo.decode(info.encode())); + } + + @Test + public void shouldEncodeAndDecodeVersion9() { + final AssignmentInfo info = + new AssignmentInfo(9, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + final AssignmentInfo expectedInfo = + new AssignmentInfo(9, LATEST_SUPPORTED_VERSION, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + assertEquals(expectedInfo, AssignmentInfo.decode(info.encode())); + } + + @Test + public void shouldEncodeAndDecodeVersion10() { + final AssignmentInfo info = + new AssignmentInfo(10, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + final AssignmentInfo expectedInfo = + new AssignmentInfo(10, LATEST_SUPPORTED_VERSION, activeTasks, standbyTasks, activeAssignment, standbyAssignment, 2); + assertEquals(expectedInfo, AssignmentInfo.decode(info.encode())); + } + + @Test + public void shouldEncodeAndDecodeVersion10WithNamedTopologies() { + final AssignmentInfo info = + new AssignmentInfo(10, LATEST_SUPPORTED_VERSION, NAMED_ACTIVE_TASKS, NAMED_STANDBY_TASKS, activeAssignment, standbyAssignment, 2); + assertEquals(info, AssignmentInfo.decode(info.encode())); + } + + @Test + public void shouldNotEncodeAndDecodeNamedTopologiesWithOlderVersion() { + final AssignmentInfo info = + new AssignmentInfo(MIN_NAMED_TOPOLOGY_VERSION - 1, LATEST_SUPPORTED_VERSION, NAMED_ACTIVE_TASKS, NAMED_STANDBY_TASKS, activeAssignment, standbyAssignment, 2); + assertThrows(TaskAssignmentException.class, () -> AssignmentInfo.decode(info.encode())); + } + @Test public void shouldEncodeAndDecodeSmallerCommonlySupportedVersion() { final int usedVersion = 5; diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentTestUtils.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentTestUtils.java index 0338fa838c547..38669da0895e7 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentTestUtils.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/AssignmentTestUtils.java @@ -87,10 +87,14 @@ public final class AssignmentTestUtils { public static final TaskId TASK_2_2 = new TaskId(2, 2); public static final TaskId TASK_2_3 = new TaskId(2, 3); - public static final TaskId NAMED_TASK_0_0 = new TaskId(0, 0, "topology0"); - public static final TaskId NAMED_TASK_0_1 = new TaskId(0, 1, "topology0"); - public static final TaskId NAMED_TASK_1_0 = new TaskId(1, 0, "topology1"); - public static final TaskId NAMED_TASK_1_1 = new TaskId(1, 1, "topology1"); + public static final TaskId NAMED_TASK_T0_0_0 = new TaskId(0, 0, "topology0"); + public static final TaskId NAMED_TASK_T0_0_1 = new TaskId(0, 1, "topology0"); + public static final TaskId NAMED_TASK_T0_1_0 = new TaskId(1, 0, "topology0"); + public static final TaskId NAMED_TASK_T0_1_1 = new TaskId(1, 1, "topology0"); + public static final TaskId NAMED_TASK_T1_0_0 = new TaskId(0, 0, "topology1"); + public static final TaskId NAMED_TASK_T1_0_1 = new TaskId(0, 1, "topology1"); + public static final TaskId NAMED_TASK_T2_0_0 = new TaskId(0, 0, "topology2"); + public static final TaskId NAMED_TASK_T2_2_0 = new TaskId(2, 0, "topology2"); public static final Subtopology SUBTOPOLOGY_0 = new Subtopology(0, null); public static final Subtopology SUBTOPOLOGY_1 = new Subtopology(1, null); diff --git a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java index f81c0cf2fec11..dd65196391578 100644 --- a/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/processor/internals/assignment/SubscriptionInfoTest.java @@ -17,6 +17,8 @@ package org.apache.kafka.streams.processor.internals.assignment; import java.util.Map; + +import org.apache.kafka.streams.errors.TaskAssignmentException; import org.apache.kafka.streams.processor.TaskId; import org.apache.kafka.streams.processor.internals.Task; import org.junit.Test; @@ -28,6 +30,14 @@ import static org.apache.kafka.common.utils.Utils.mkEntry; import static org.apache.kafka.common.utils.Utils.mkMap; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_0_1; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_1_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T0_1_1; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T1_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T1_0_1; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T2_0_0; +import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.NAMED_TASK_T2_2_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_0_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_0_1; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_1_0; @@ -35,6 +45,7 @@ import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.TASK_2_0; import static org.apache.kafka.streams.processor.internals.assignment.AssignmentTestUtils.UUID_1; import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION; +import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.MIN_NAMED_TOPOLOGY_VERSION; import static org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo.MIN_VERSION_OFFSET_SUM_SUBSCRIPTION; import static org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo.UNKNOWN_OFFSET_SUM; import static org.hamcrest.CoreMatchers.is; @@ -60,6 +71,17 @@ public class SubscriptionInfoTest { mkEntry(TASK_2_0, 10L) ); + private static final Map NAMED_TASK_OFFSET_SUMS = mkMap( + mkEntry(NAMED_TASK_T0_0_0, Task.LATEST_OFFSET), + mkEntry(NAMED_TASK_T0_0_1, Task.LATEST_OFFSET), + mkEntry(NAMED_TASK_T0_1_0, 5L), + mkEntry(NAMED_TASK_T0_1_1, 10_000L), + mkEntry(NAMED_TASK_T1_0_0, Task.LATEST_OFFSET), + mkEntry(NAMED_TASK_T1_0_1, 0L), + mkEntry(NAMED_TASK_T2_0_0, 10L), + mkEntry(NAMED_TASK_T2_2_0, 5L) + ); + private final static String IGNORED_USER_ENDPOINT = "ignoredUserEndpoint:80"; private static final byte IGNORED_UNIQUE_FIELD = (byte) 0; private static final int IGNORED_ERROR_CODE = 0; @@ -389,6 +411,28 @@ public void shouldEncodeAndDecodeVersion9() { assertThat(info, is(SubscriptionInfo.decode(info.encode()))); } + @Test + public void shouldEncodeAndDecodeVersion10() { + final SubscriptionInfo info = + new SubscriptionInfo(10, LATEST_SUPPORTED_VERSION, UUID_1, "localhost:80", TASK_OFFSET_SUMS, IGNORED_UNIQUE_FIELD, IGNORED_ERROR_CODE); + assertThat(info, is(SubscriptionInfo.decode(info.encode()))); + } + + @Test + public void shouldEncodeAndDecodeVersion10WithNamedTopologies() { + final SubscriptionInfo info = + new SubscriptionInfo(10, LATEST_SUPPORTED_VERSION, UUID_1, "localhost:80", NAMED_TASK_OFFSET_SUMS, IGNORED_UNIQUE_FIELD, IGNORED_ERROR_CODE); + assertThat(info, is(SubscriptionInfo.decode(info.encode()))); + } + + @Test + public void shouldThrowIfAttemptingToUseNamedTopologiesWithOlderVersion() { + assertThrows( + TaskAssignmentException.class, + () -> new SubscriptionInfo(MIN_NAMED_TOPOLOGY_VERSION - 1, LATEST_SUPPORTED_VERSION, UUID_1, "localhost:80", NAMED_TASK_OFFSET_SUMS, IGNORED_UNIQUE_FIELD, IGNORED_ERROR_CODE) + ); + } + private static ByteBuffer encodeFutureVersion() { final ByteBuffer buf = ByteBuffer.allocate(4 /* used version */ + 4 /* supported version */); diff --git a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java index 2a16547f780bf..c8a320f8cfc09 100644 --- a/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java +++ b/streams/src/test/java/org/apache/kafka/streams/state/internals/StreamThreadStateStoreProviderTest.java @@ -153,7 +153,7 @@ public void before() { final ProcessorTopology processorTopology = internalTopologyBuilder.buildTopology(); tasks = new HashMap<>(); - stateDirectory = new StateDirectory(streamsConfig, new MockTime(), true); + stateDirectory = new StateDirectory(streamsConfig, new MockTime(), true, false); taskOne = createStreamsTask( streamsConfig, diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java index 63e6a278da06d..779a84f56a11f 100644 --- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java +++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java @@ -419,7 +419,7 @@ private void setupTopology(final InternalTopologyBuilder builder, final boolean createStateDirectory = processorTopology.hasPersistentLocalStore() || (globalTopology != null && globalTopology.hasPersistentGlobalStore()); - stateDirectory = new StateDirectory(streamsConfig, mockWallClockTime, createStateDirectory); + stateDirectory = new StateDirectory(streamsConfig, mockWallClockTime, createStateDirectory, builder.hasNamedTopologies()); } private void setupGlobalTask(final Time mockWallClockTime, diff --git a/tests/kafkatest/tests/streams/streams_upgrade_test.py b/tests/kafkatest/tests/streams/streams_upgrade_test.py index 9aff673349c8d..cb5375de02ace 100644 --- a/tests/kafkatest/tests/streams/streams_upgrade_test.py +++ b/tests/kafkatest/tests/streams/streams_upgrade_test.py @@ -474,12 +474,13 @@ def do_rolling_bounce(self, processor, counter, current_generation): monitors[first_other_processor] = first_other_monitor monitors[second_other_processor] = second_other_monitor - end_of_upgrade_message = "Sent a version 9 subscription and group.s latest commonly supported version is 10 (successful version probing and end of rolling upgrade). Upgrading subscription metadata version to 10 for next rebalance." + version_probing_message = "Sent a version 10 subscription and got version 10 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 10 and trigger new rebalance.", + end_of_upgrade_message = "Sent a version 10 subscription and group.s latest commonly supported version is 11 (successful version probing and end of rolling upgrade). Upgrading subscription metadata version to 11 for next rebalance." end_of_upgrade_error_message = "Could not detect 'successful version probing and end of rolling upgrade' at upgraded node " followup_rebalance_message = "Triggering the followup rebalance scheduled for 0 ms." followup_rebalance_error_message = "Could not detect 'Triggering followup rebalance' at node " if len(self.old_processors) > 0: - log_monitor.wait_until("Sent a version 10 subscription and got version 9 assignment back (successful version probing). Downgrade subscription metadata to commonly supported version 9 and trigger new rebalance.", + log_monitor.wait_until(version_probing_message, timeout_sec=60, err_msg="Could not detect 'successful version probing' at upgrading node " + str(node.account)) log_monitor.wait_until(followup_rebalance_message,