Skip to content

KAFKA-6145: KIP-441: Add TaskAssignor class config #8541

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vvcephei merged 12 commits into from
Apr 28, 2020
Merged

KAFKA-6145: KIP-441: Add TaskAssignor class config #8541

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
97228a5
KAFKA-6145: KIP-441: Add TaskAssignor class config
Apr 21, 2020
fed2506
restore LagFetchIntegrationTest
Apr 24, 2020
b7373de
cr feedback
Apr 24, 2020
deefb58
cr comments and fixing system tests in progress
Apr 24, 2020
77fbb41
fix system tests
Apr 24, 2020
9015be8
fix tests
Apr 24, 2020
126afd1
revert unnecessary change
Apr 24, 2020
98e317e
Update streams/src/test/java/org/apache/kafka/streams/processor/inter…
vvcephei Apr 27, 2020
e3bf615
Update streams/src/test/java/org/apache/kafka/streams/processor/inter…
vvcephei Apr 27, 2020
e077be2
cr feedback
Apr 27, 2020
d22de81
cr feedback
Apr 28, 2020
d58f62d
rename 'followup' to 'probing'
Apr 28, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -236,8 +236,10 @@ subprojects {
def logStreams = new HashMap<String, FileOutputStream>()
beforeTest { TestDescriptor td ->
def tid = testId(td)
// truncate the file name if it's too long
def logFile = new File(
"${projectDir}/build/reports/testOutput/${tid}.test.stdout")
"${projectDir}/build/reports/testOutput/${tid.substring(0, Math.min(tid.size(),240))}.test.stdout"

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Necessary because the test name that JUnit generates for the parameterized StreamsPartitionAssignorTest is slightly too long. I have no way to shorten it because the thing that pushes it over is the fact that there are two package names in the parameterized method name, and there's no control over the format of the test name itself. So, I decided just to truncate the file name instead, which is almost certainly still unique for pretty much any test.

@vvcephei vvcephei Apr 24, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The only alternative I can think of is to parameterize the "short name" of the TaskAssignor, which seems kind of wacky.

Also, worth noting the impact of truncation is nothing if the file name is still unique. If the name is shared between two tests, then the impact is still nothing if both tests pass. The only observable effect is that if one or both tests fail, their logs would get combined. It seems like we can afford just to defer this problem until it happens, if ever.

)
logFile.parentFile.mkdirs()
logFiles.put(tid, logFile)
logStreams.put(tid, new FileOutputStream(logFile))
Expand Down
9 changes: 9 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1146,4 +1146,13 @@ public Set<Characteristics> characteristics() {
}
};
}

@SafeVarargs
public static <E> Set<E> union(final Supplier<Set<E>> constructor, final Set<E>... set) {

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I've been wanting this for a while, so I just decided to add it.

@cadonna cadonna Apr 27, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

req: Please add unit tests for this method

final Set<E> result = constructor.get();
for (final Set<E> s : set) {
result.addAll(s);
}
return result;
}
}
4 changes: 2 additions & 2 deletions clients/src/test/java/org/apache/kafka/test/TestUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,9 +361,9 @@ public static void waitForCondition(final TestCondition testCondition, final lon
* avoid transient failures due to slow or overloaded machines.
*/
public static void waitForCondition(final TestCondition testCondition, final long maxWaitMs, Supplier<String> conditionDetailsSupplier) throws InterruptedException {
String conditionDetailsSupplied = conditionDetailsSupplier != null ? conditionDetailsSupplier.get() : null;
String conditionDetails = conditionDetailsSupplied != null ? conditionDetailsSupplied : "";
retryOnExceptionWithTimeout(maxWaitMs, () -> {
String conditionDetailsSupplied = conditionDetailsSupplier != null ? conditionDetailsSupplier.get() : null;
String conditionDetails = conditionDetailsSupplied != null ? conditionDetailsSupplied : "";

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This is pointless unless we evaluate it inside the lambda.

assertThat("Condition not met within timeout " + maxWaitMs + ". " + conditionDetails,
testCondition.conditionMet());
});
Expand Down
33 changes: 14 additions & 19 deletions .../src/main/java/org/apache/kafka/streams/processor/internals/StreamsPartitionAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,7 @@
import org.apache.kafka.streams.processor.internals.assignment.AssignorError;
import org.apache.kafka.streams.processor.internals.assignment.ClientState;
import org.apache.kafka.streams.processor.internals.assignment.CopartitionedTopicsEnforcer;
import org.apache.kafka.streams.processor.internals.assignment.HighAvailabilityTaskAssignor;
import org.apache.kafka.streams.processor.internals.assignment.StickyTaskAssignor;
import org.apache.kafka.streams.processor.internals.assignment.PriorTaskAssignor;
import org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo;
import org.apache.kafka.streams.processor.internals.assignment.TaskAssignor;
import org.apache.kafka.streams.state.HostInfo;
Expand All @@ -64,6 +63,7 @@
import java.util.UUID;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static java.util.UUID.randomUUID;
Expand Down Expand Up @@ -171,7 +171,7 @@ public String toString() {
private CopartitionedTopicsEnforcer copartitionedTopicsEnforcer;
private RebalanceProtocol rebalanceProtocol;

private boolean highAvailabilityEnabled;
private Supplier<TaskAssignor> taskAssignor;

/**
* We need to have the PartitionAssignor and its StreamThread to be mutually accessible since the former needs
Expand Down Expand Up @@ -201,7 +201,7 @@ public void configure(final Map<String, ?> configs) {
internalTopicManager = assignorConfiguration.getInternalTopicManager();
copartitionedTopicsEnforcer = assignorConfiguration.getCopartitionedTopicsEnforcer();
rebalanceProtocol = assignorConfiguration.rebalanceProtocol();
highAvailabilityEnabled = assignorConfiguration.isHighAvailabilityEnabled();
taskAssignor = assignorConfiguration::getTaskAssignor;
}

@Override
Expand Down Expand Up @@ -713,23 +713,18 @@ private boolean assignTasksToClients(final Set<String> allSourceTopics,
allTasks, clientStates, numStandbyReplicas());

final TaskAssignor taskAssignor;
if (highAvailabilityEnabled) {
if (lagComputationSuccessful) {
taskAssignor = new HighAvailabilityTaskAssignor(
clientStates,
allTasks,
statefulTasks,
assignmentConfigs);
} else {
log.info("Failed to fetch end offsets for changelogs, will return previous assignment to clients and "
+ "trigger another rebalance to retry.");
setAssignmentErrorCode(AssignorError.REBALANCE_NEEDED.code());
taskAssignor = new StickyTaskAssignor(clientStates, allTasks, statefulTasks, assignmentConfigs, true);
}
if (!lagComputationSuccessful) {
Comment thread
vvcephei marked this conversation as resolved.
Outdated
log.info("Failed to fetch end offsets for changelogs, will return previous assignment to clients and "
+ "trigger another rebalance to retry.");
setAssignmentErrorCode(AssignorError.REBALANCE_NEEDED.code());
taskAssignor = new PriorTaskAssignor();

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just to clarify everyone's roles, I added a new assignor whose only behavior is to return all previously owned tasks, and then assign any unowned tasks.

} else {
taskAssignor = new StickyTaskAssignor(clientStates, allTasks, statefulTasks, assignmentConfigs, false);
taskAssignor = this.taskAssignor.get();
}
Comment thread
vvcephei marked this conversation as resolved.
Outdated
final boolean followupRebalanceNeeded = taskAssignor.assign();
final boolean followupRebalanceNeeded = taskAssignor.assign(clientStates,

@ableegoldman ableegoldman Apr 28, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should probably rename this to probingRebalanceRequired or so on, see comment on FallbackPriorTaskAssignor

@vvcephei vvcephei Apr 28, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, seems legit.

allTasks,
statefulTasks,
assignmentConfigs);

log.info("Assigned tasks to clients as {}{}.",
Utils.NL, clientStates.entrySet().stream().map(Map.Entry::toString).collect(Collectors.joining(Utils.NL)));
Expand Down
22 changes: 15 additions & 7 deletions ...n/java/org/apache/kafka/streams/processor/internals/assignment/AssignorConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import org.apache.kafka.common.config.ConfigException;
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.StreamsConfig.InternalConfig;
import org.apache.kafka.streams.internals.QuietStreamsConfig;
Expand All @@ -41,8 +42,8 @@
import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.LATEST_SUPPORTED_VERSION;

public final class AssignorConfiguration {
public static final String HIGH_AVAILABILITY_ENABLED_CONFIG = "internal.high.availability.enabled";
private final boolean highAvailabilityEnabled;
public static final String INTERNAL_TASK_ASSIGNOR_CLASS = "internal.task.assignor.class";

@ableegoldman ableegoldman Apr 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should we put this with the other Streams internal configs? And/or follow the pattern of prefix+suffixing with __ ?

@vvcephei vvcephei Apr 24, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, yeah, good idea.

@vvcephei vvcephei Apr 24, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, I moved it to org.apache.kafka.streams.StreamsConfig.InternalConfig#INTERNAL_TASK_ASSIGNOR_CLASS. I made an ad-hoc decision not to add the underscores, though, because this config is different than the other internal configs. I added comments to InternalConfig to explain the difference.

private final String taskAssignorClass;

private final String logPrefix;
private final Logger log;
Expand Down Expand Up @@ -162,11 +163,11 @@ public AssignorConfiguration(final Map<String, ?> configs) {
copartitionedTopicsEnforcer = new CopartitionedTopicsEnforcer(logPrefix);

{
final Object o = configs.get(HIGH_AVAILABILITY_ENABLED_CONFIG);
final String o = (String) configs.get(INTERNAL_TASK_ASSIGNOR_CLASS);
if (o == null) {
highAvailabilityEnabled = false;
taskAssignorClass = HighAvailabilityTaskAssignor.class.getName();
} else {
highAvailabilityEnabled = (Boolean) o;
taskAssignorClass = o;
}
}
}
Expand Down Expand Up @@ -328,8 +329,15 @@ public AssignmentConfigs getAssignmentConfigs() {
return assignmentConfigs;
}

public boolean isHighAvailabilityEnabled() {
return highAvailabilityEnabled;
public TaskAssignor getTaskAssignor() {
try {
return Utils.newInstance(taskAssignorClass, TaskAssignor.class);
} catch (final ClassNotFoundException e) {
throw new IllegalArgumentException(
"Expected an instantiable class name for " + INTERNAL_TASK_ASSIGNOR_CLASS,
e
);
}
}

public static class AssignmentConfigs {
Expand Down
22 changes: 21 additions & 1 deletion ...ms/src/main/java/org/apache/kafka/streams/processor/internals/assignment/ClientState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,10 @@
import java.util.Set;
import java.util.UUID;

import static java.util.Collections.emptyMap;
import static java.util.Collections.unmodifiableMap;
import static java.util.Collections.unmodifiableSet;
import static org.apache.kafka.common.utils.Utils.union;
import static org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo.UNKNOWN_OFFSET_SUM;

public class ClientState {
Expand Down Expand Up @@ -86,6 +90,22 @@ private ClientState(final Set<TaskId> activeTasks,
this.capacity = capacity;
}

public ClientState(final Set<TaskId> previousActiveTasks,
Comment thread
vvcephei marked this conversation as resolved.
final Set<TaskId> previousStandbyTasks,
final Map<TaskId, Long> taskLagTotals,
final int capacity) {
Comment on lines 93 to 96

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This constructor is currently only used in tests, but I'm planning a follow-on refactor that would actually use it from the StickyTaskAssignor as well.

@ableegoldman ableegoldman Apr 28, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm having a failure of imagination to see why any task assignor would ever be creating ClientState objects, but I eagerly wait to be enlightened.

activeTasks = new HashSet<>();
standbyTasks = new HashSet<>();
assignedTasks = new HashSet<>();
prevActiveTasks = unmodifiableSet(new HashSet<>(previousActiveTasks));
prevStandbyTasks = unmodifiableSet(new HashSet<>(previousStandbyTasks));
prevAssignedTasks = unmodifiableSet(union(HashSet::new, previousActiveTasks, previousStandbyTasks));
ownedPartitions = emptyMap();
taskOffsetSums = emptyMap();
this.taskLagTotals = unmodifiableMap(taskLagTotals);
this.capacity = capacity;
}

public ClientState copy() {
return new ClientState(
new HashSet<>(activeTasks),
Expand Down Expand Up @@ -258,7 +278,7 @@ boolean hasUnfulfilledQuota(final int tasksPerThread) {
}

boolean hasMoreAvailableCapacityThan(final ClientState other) {
if (this.capacity <= 0) {
if (capacity <= 0) {
throw new IllegalStateException("Capacity of this ClientState must be greater than 0.");
}

Expand Down
56 changes: 30 additions & 26 deletions ...org/apache/kafka/streams/processor/internals/assignment/HighAvailabilityTaskAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,49 +16,50 @@
*/
package org.apache.kafka.streams.processor.internals.assignment;

import static org.apache.kafka.streams.processor.internals.assignment.AssignmentUtils.taskIsCaughtUpOnClientOrNoCaughtUpClientsExist;
import static org.apache.kafka.streams.processor.internals.assignment.RankedClient.buildClientRankingsByTask;
import static org.apache.kafka.streams.processor.internals.assignment.RankedClient.tasksToCaughtUpClients;
import static org.apache.kafka.streams.processor.internals.assignment.TaskMovement.assignTaskMovements;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedMap;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.Set;
import static org.apache.kafka.streams.processor.internals.assignment.AssignmentUtils.taskIsCaughtUpOnClientOrNoCaughtUpClientsExist;
import static org.apache.kafka.streams.processor.internals.assignment.RankedClient.buildClientRankingsByTask;
import static org.apache.kafka.streams.processor.internals.assignment.RankedClient.tasksToCaughtUpClients;
import static org.apache.kafka.streams.processor.internals.assignment.TaskMovement.assignTaskMovements;
Comment thread
vvcephei marked this conversation as resolved.

public class HighAvailabilityTaskAssignor implements TaskAssignor {
private static final Logger log = LoggerFactory.getLogger(HighAvailabilityTaskAssignor.class);

private final Map<UUID, ClientState> clientStates;
private final Map<UUID, Integer> clientsToNumberOfThreads;
private final SortedSet<UUID> sortedClients;
private Map<UUID, ClientState> clientStates;

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

All these fields have to be non-final now, because we're setting them in assign instead of the constructor.

private Map<UUID, Integer> clientsToNumberOfThreads;
private SortedSet<UUID> sortedClients;

private final Set<TaskId> allTasks;
private final SortedSet<TaskId> statefulTasks;
private final SortedSet<TaskId> statelessTasks;
private Set<TaskId> allTasks;
private SortedSet<TaskId> statefulTasks;
private SortedSet<TaskId> statelessTasks;

private final AssignmentConfigs configs;
private AssignmentConfigs configs;

private final SortedMap<TaskId, SortedSet<RankedClient>> statefulTasksToRankedCandidates;
private final Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients;
private SortedMap<TaskId, SortedSet<RankedClient>> statefulTasksToRankedCandidates;
private Map<TaskId, SortedSet<UUID>> tasksToCaughtUpClients;

public HighAvailabilityTaskAssignor(final Map<UUID, ClientState> clientStates,
final Set<TaskId> allTasks,
final Set<TaskId> statefulTasks,
final AssignmentConfigs configs) {
@Override
public boolean assign(final Map<UUID, ClientState> clientStates,
final Set<TaskId> allTasks,
final Set<TaskId> statefulTasks,
final AssignmentConfigs configs) {
this.configs = configs;
this.clientStates = clientStates;
this.allTasks = allTasks;
Expand All @@ -77,10 +78,8 @@ public HighAvailabilityTaskAssignor(final Map<UUID, ClientState> clientStates,
statefulTasksToRankedCandidates =
buildClientRankingsByTask(statefulTasks, clientStates, configs.acceptableRecoveryLag);
tasksToCaughtUpClients = tasksToCaughtUpClients(statefulTasksToRankedCandidates);
}

@Override
public boolean assign() {

if (shouldUsePreviousAssignment()) {
assignPreviousTasksToClientStates();
return false;
Expand All @@ -95,6 +94,11 @@ public boolean assign() {

assignStatelessActiveTasks();

log.info("Decided on assignment: " +

@vvcephei vvcephei Apr 23, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I found this log useful while debugging the integration tests. WDYT about keeping it?

@ableegoldman ableegoldman Apr 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm all for useful logging 👍

clientStates +
" with " +
(followupRebalanceNeeded ? "" : "no") +
" followup rebalance.");
return followupRebalanceNeeded;
}

Expand Down
40 changes: 40 additions & 0 deletions .../main/java/org/apache/kafka/streams/processor/internals/assignment/PriorTaskAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,40 @@
/*
* 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.processor.internals.assignment;

import org.apache.kafka.streams.processor.TaskId;
import org.apache.kafka.streams.processor.internals.assignment.AssignorConfiguration.AssignmentConfigs;

import java.util.Map;
import java.util.Set;
import java.util.UUID;

public class PriorTaskAssignor implements TaskAssignor {
private final StickyTaskAssignor delegate;

public PriorTaskAssignor() {
delegate = new StickyTaskAssignor(true);

@vvcephei vvcephei Apr 23, 2020 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The StickyTaskAssignor is capable of satisfying the PriorTaskAssignor's contract, so we can just delegate to it. The important thing is that we now have two separately defined contracts:

  1. return all previous tasks and assign the rest (PriorTaskAssignor)
  2. strike a balance between stickiness and balance (StickyTaskAssignor)

The fact that the implementation is shared is an ... implementation detail.

@ableegoldman ableegoldman Apr 24, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks for the improvement, this feels a lot nicer

}

@Override
public boolean assign(final Map<UUID, ClientState> clients,
final Set<TaskId> allTaskIds,
final Set<TaskId> standbyTaskIds,
final AssignmentConfigs configs) {
return delegate.assign(clients, allTaskIds, standbyTaskIds, configs);
}
}
Loading