Skip to content

KAFKA-10079: improve thread-level stickiness #8775

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 23 commits into from
Jun 10, 2020
Merged

KAFKA-10079: improve thread-level stickiness #8775

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
23 commits
Select commit Hold shift + click to select a range
9140097
WIP
ableegoldman Jun 2, 2020
5083889
implemented stickiness
ableegoldman Jun 2, 2020
be11863
compiling tests
ableegoldman Jun 2, 2020
dc296fc
improve stickiness to the max
ableegoldman Jun 2, 2020
4b8e96c
debugging last test
ableegoldman Jun 2, 2020
8300d49
use sorted set for test
ableegoldman Jun 2, 2020
d40dd5a
checkstyle
ableegoldman Jun 2, 2020
5cb6f40
add unit tests
ableegoldman Jun 2, 2020
72e64ef
filter for statefulness in SPA
ableegoldman Jun 2, 2020
acb7a4c
fixing up tests
ableegoldman Jun 2, 2020
f580e63
bump log to INFO
ableegoldman Jun 3, 2020
7c18e59
pull sorting into SPA
ableegoldman Jun 4, 2020
96a6c23
checkstyle
ableegoldman Jun 4, 2020
d26d462
use UNKNOWN_OFFSET_SUM, always initialize lag
ableegoldman Jun 4, 2020
6df10ca
remove no longer relevant new test
ableegoldman Jun 4, 2020
fc98357
fix/rename counter
ableegoldman Jun 4, 2020
b5cfd40
some cleanup
ableegoldman Jun 4, 2020
f83baf9
fixindentation
ableegoldman Jun 4, 2020
f0c5a19
simplify
ableegoldman Jun 4, 2020
ee384a9
fix bug caught in test
ableegoldman Jun 4, 2020
168051c
Adding unit tests
ableegoldman Jun 8, 2020
6b0f6fb
checkstyle
ableegoldman Jun 8, 2020
e921175
remove unused topic partitions
ableegoldman Jun 8, 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
9 changes: 8 additions & 1 deletion .../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 @@ -70,6 +70,7 @@
import java.util.function.Supplier;
import java.util.stream.Collectors;

import static java.util.Comparator.comparingLong;
import static java.util.UUID.randomUUID;
import static org.apache.kafka.streams.processor.internals.ClientUtils.fetchEndOffsets;
import static org.apache.kafka.streams.processor.internals.assignment.StreamsAssignmentProtocolVersions.EARLIEST_PROBEABLE_VERSION;
Expand Down Expand Up @@ -1099,7 +1100,7 @@ static Map<String, List<TaskId>> assignTasksToThreads(final Collection<TaskId> s
final List<TaskId> threadAssignment = assignment.get(consumer);

int i = 0;
for (final TaskId task : state.previousTasksForConsumer(consumer)) {
for (final TaskId task : getPreviousTasksByLag(state, consumer)) {
if (unassignedStatefulTasks.contains(task)) {
if (i < minStatefulTasksPerThread) {
threadAssignment.add(task);
Expand Down Expand Up @@ -1179,6 +1180,12 @@ static Map<String, List<TaskId>> assignTasksToThreads(final Collection<TaskId> s
return assignment;
}

private static SortedSet<TaskId> getPreviousTasksByLag(final ClientState state, final String consumer) {
final SortedSet<TaskId> prevTasksByLag = new TreeSet<>(comparingLong(state::lagFor).thenComparing(TaskId::compareTo));
prevTasksByLag.addAll(state.previousTasksForConsumer(consumer));
return prevTasksByLag;
}

private void validateMetadataVersions(final int receivedAssignmentMetadataVersion,
final int latestCommonlySupportedVersion) {

Expand Down
29 changes: 11 additions & 18 deletions ...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 @@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams.processor.internals.assignment;

import java.util.SortedSet;
import java.util.stream.Collectors;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.streams.processor.TaskId;
Expand All @@ -38,7 +37,6 @@
import static java.util.Collections.unmodifiableMap;
import static java.util.Collections.unmodifiableSet;
import static java.util.Comparator.comparing;
import static java.util.Comparator.comparingLong;
import static org.apache.kafka.common.utils.Utils.union;
import static org.apache.kafka.streams.processor.internals.assignment.SubscriptionInfo.UNKNOWN_OFFSET_SUM;

Expand Down Expand Up @@ -302,14 +300,19 @@ public void computeTaskLags(final UUID uuid, final Map<TaskId, Long> allTaskEndO
* @return end offset sum - offset sum
* Task.LATEST_OFFSET if this was previously an active running task on this client
*/
long lagFor(final TaskId task) {
final Long totalLag = taskLagTotals.get(task);
public long lagFor(final TaskId task) {
final Long totalLag;
if (taskLagTotals.isEmpty()) {
// If we couldn't compute the task lags due to failure to fetch offsets, just return a flat constant
totalLag = 0L;

@vvcephei vvcephei Jun 4, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Is this the right constant to represent "we don't know the lag"? Or did I mistake how this is going to be used?

@ableegoldman ableegoldman Jun 4, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

The value itself doesn't matter, just that it's constant across all tasks.

But I'm guessing you meant, why not use the existing UNKNOWN_OFFSET_SUM sentinel, in which case the answer is probably just that I forgot about it. Anyway I did a slight additional refactoring beyond this, just fyi: instead of skipping the lag computation when we fail to fetch offsets, we now always initialize the lags and just pass in the UNKNOWN_OFFSET_SUM for all stateful tasks when the offset fetch fails.

} else {
totalLag = taskLagTotals.get(task);
}

if (totalLag == null) {
throw new IllegalStateException("Tried to lookup lag for unknown task " + task);
} else {
return totalLag;
}
return totalLag;
}

public Set<TaskId> statefulActiveTasks() {
Expand All @@ -320,18 +323,8 @@ public Set<TaskId> statelessActiveTasks() {
return activeTasks.stream().filter(task -> !isStateful(task)).collect(Collectors.toSet());
}

// Return a list of that consumer's previous tasks in increasing lag order
public SortedSet<TaskId> previousTasksForConsumer(final String memberId) {
final Set<TaskId> prevTasks = consumerToPreviousStatefulTaskIds.get(memberId);

// If we were unable to fetch the end offsets and could not compute lags, just return in task order
if (taskLagTotals.isEmpty()) {
return new TreeSet<>(prevTasks);
} else {
final SortedSet<TaskId> prevTasksByLag = new TreeSet<>(comparingLong(this::lagFor).thenComparing(TaskId::compareTo));
prevTasksByLag.addAll(prevTasks);
return prevTasksByLag;
}
public Set<TaskId> previousTasksForConsumer(final String memberId) {
return consumerToPreviousStatefulTaskIds.get(memberId);
}

boolean hasUnfulfilledQuota(final int tasksPerThread) {
Expand Down
38 changes: 14 additions & 24 deletions ...rc/test/java/org/apache/kafka/streams/processor/internals/assignment/ClientStateTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,6 @@
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.common.utils.Utils.mkSortedSet;
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_0_2;
Expand All @@ -46,7 +45,6 @@
import static org.junit.Assert.assertTrue;

public class ClientStateTest {

private final ClientState client = new ClientState(1);
private final ClientState zeroCapacityClient = new ClientState(0);

Expand Down Expand Up @@ -324,40 +322,22 @@ public void shouldReturnPreviousStatefulTasksForConsumer() {
)
);

assertThat(client.previousTasksForConsumer("c1"), equalTo(mkSortedSet(TASK_0_1)));
assertThat(client.previousTasksForConsumer("c2"), equalTo(mkSortedSet(TASK_0_2)));
assertThat(client.previousTasksForConsumer("c1"), equalTo(mkSet(TASK_0_1)));
assertThat(client.previousTasksForConsumer("c2"), equalTo(mkSet(TASK_0_2)));
assertTrue(client.previousTasksForConsumer("c3").isEmpty());
}

@Test
public void shouldReturnPreviousStatefulTasksForConsumerWhenLagIsNotComputed() {
client.addPreviousTasksAndOffsetSums("c1", Collections.singletonMap(TASK_0_1, 1000L));
client.initializePrevTasks(Collections.emptyMap());

assertThat(client.previousTasksForConsumer("c1"), equalTo(mkSortedSet(TASK_0_1)));
}

@Test
public void shouldReturnPreviousStatefulTasksForConsumerInIncreasingLagOrder() {
public void shouldReturnPreviousTasksForConsumer() {
client.addPreviousTasksAndOffsetSums("c1", mkMap(
mkEntry(TASK_0_1, 100L),
mkEntry(TASK_0_2, 0L),
mkEntry(TASK_0_3, Task.LATEST_OFFSET)
));

client.computeTaskLags(
UUID_1,
mkMap(
mkEntry(TASK_0_1, 1_000L),
mkEntry(TASK_0_2, 1_000L),
mkEntry(TASK_0_3, 1_000L)

)
);

client.initializePrevTasks(Collections.emptyMap());

assertThat(client.previousTasksForConsumer("c1"), equalTo(mkSortedSet(TASK_0_3, TASK_0_2, TASK_0_1)));
assertThat(client.previousTasksForConsumer("c1"), equalTo(mkSet(TASK_0_3, TASK_0_2, TASK_0_1)));
}

@Test
Expand Down Expand Up @@ -434,6 +414,16 @@ public void shouldThrowIllegalStateExceptionIfTaskLagsMapIsNotEmpty() {
assertThrows(IllegalStateException.class, () -> client.computeTaskLags(null, allTaskEndOffsetSums));
}

@Test
public void shouldReturnZeroForAllTasksIfLagNotComputed() {
client.addPreviousTasksAndOffsetSums("c1", Collections.singletonMap(TASK_0_1, 0L));
client.addPreviousTasksAndOffsetSums("c2", Collections.singletonMap(TASK_0_2, Task.LATEST_OFFSET));
client.addPreviousTasksAndOffsetSums("c3", Collections.singletonMap(TASK_0_3, 500L));
assertThat(client.lagFor(TASK_0_1), equalTo(0L));
assertThat(client.lagFor(TASK_0_2), equalTo(0L));
assertThat(client.lagFor(TASK_0_3), equalTo(0L));
}

@Test
public void shouldThrowIllegalStateExceptionOnLagForUnknownTask() {
final Map<TaskId, Long> taskOffsetSums = Collections.singletonMap(TASK_0_1, 0L);
Expand Down