Skip to content
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

Unsubscribe on partition reassignment #1021

Merged
merged 5 commits into from
May 16, 2024
Merged

Unsubscribe on partition reassignment #1021

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
32f9bba
Unsubscribe on partition reassignment
akrambek May 10, 2024
ac681c9
Use dataEx to request for updated offset fetch
akrambek May 11, 2024
029e670
Handle empty partition assignment and make assignments sticky
akrambek May 14, 2024
26c8266
Add more spec to cover edge cases
akrambek May 14, 2024
69d0c53
Refactor code
akrambek May 15, 2024
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
1 change: 1 addition & 0 deletions ...ka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/KafkaConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ public class KafkaConfiguration extends Configuration
{
public static final boolean DEBUG = Boolean.getBoolean("zilla.binding.kafka.debug");
public static final boolean DEBUG_PRODUCE = DEBUG || Boolean.getBoolean("zilla.binding.kafka.debug.produce");
public static final boolean DEBUG_CONSUMER = DEBUG || Boolean.getBoolean("zilla.binding.kafka.debug.consumer");

public static final String KAFKA_CLIENT_ID_DEFAULT = "zilla";

Expand Down
38 changes: 28 additions & 10 deletions ...aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheServerConsumerFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@

import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.LongFunction;
import java.util.function.LongUnaryOperator;
Expand All @@ -26,6 +28,7 @@
import org.agrona.MutableDirectBuffer;
import org.agrona.collections.IntHashSet;
import org.agrona.collections.Object2ObjectHashMap;
import org.agrona.collections.ObjectHashSet;
import org.agrona.concurrent.UnsafeBuffer;

import io.aklivity.zilla.runtime.binding.kafka.internal.KafkaBinding;
Expand Down Expand Up @@ -533,7 +536,8 @@ final class KafkaCacheServerConsumerFanout
private final long routedId;
private final long authorization;
private final ArrayList<KafkaCacheServerConsumerStream> streams;
private final Object2ObjectHashMap<String, String> members;
private final Map<String, String> members;
private final ObjectHashSet<String> memberIds;
private final Object2ObjectHashMap<String, IntHashSet> partitionsByTopic;
private final Object2ObjectHashMap<String, List<TopicPartition>> consumers;
private final Object2ObjectHashMap<String, TopicConsumer> assignments;
Expand Down Expand Up @@ -578,7 +582,8 @@ private KafkaCacheServerConsumerFanout(
this.groupId = groupId;
this.timeout = timeout;
this.streams = new ArrayList<>();
this.members = new Object2ObjectHashMap<>();
this.memberIds = new ObjectHashSet<>();
this.members = new LinkedHashMap<>();
Copy link
Contributor

Choose a reason for hiding this comment

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

Are we switching to LinkedHashMap to maintain insertion order, or for putIfAbsent?
Note: putIfAbsent is also available on Object2ObjectHashMap, inherited from Map.

Copy link
Contributor Author

Choose a reason for hiding this comment

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

to maintain insertion order

this.partitionsByTopic = new Object2ObjectHashMap<>();
this.consumers = new Object2ObjectHashMap<>();
this.assignments = new Object2ObjectHashMap<>();
Expand All @@ -590,8 +595,6 @@ private void onConsumerFanoutStreamOpening(
{
streams.add(stream);

assert !streams.isEmpty();

doConsumerInitialBegin(traceId, stream);

if (KafkaState.initialOpened(state))
Expand Down Expand Up @@ -886,7 +889,7 @@ private void onConsumerReplyFlush(
memberId = kafkaGroupFlushEx.memberId().asString();

partitionsByTopic.clear();
members.clear();
memberIds.clear();

kafkaGroupFlushEx.members().forEach(m ->
{
Expand All @@ -896,7 +899,8 @@ private void onConsumerReplyFlush(
final String consumerId = kafkaGroupMemberMetadataRO.consumerId().asString();

final String mId = m.id().asString();
members.put(mId, consumerId);
members.putIfAbsent(mId, consumerId);
memberIds.add(mId);

groupMetadata.topics().forEach(mt ->
{
Expand All @@ -907,6 +911,8 @@ private void onConsumerReplyFlush(
});
}

members.entrySet().removeIf(m -> !memberIds.contains(m.getKey()));

doPartitionAssignment(traceId, authorization);
}

Expand Down Expand Up @@ -941,6 +947,12 @@ private void onConsumerReplyData(
IntHashSet partitions = new IntHashSet();
List<TopicPartition> topicConsumers = new ArrayList<>();

if (KafkaConfiguration.DEBUG_CONSUMER)
{
System.out.printf("Subscription: MemberId - %s\n", memberId);
ta.partitions().forEach(np -> System.out.printf("%s\n", np));
}

stream.doConsumerReplyData(traceId, flags, replyPad, EMPTY_OCTETS,
ex -> ex.set((b, o, l) -> kafkaDataExRW.wrap(b, o, l)
.typeId(kafkaTypeId)
Expand Down Expand Up @@ -1089,7 +1101,6 @@ private void doMemberAssigment(
.wrap(writeBuffer, DataFW.FIELD_OFFSET_PAYLOAD, writeBuffer.capacity());

this.consumers.forEach((k, v) ->
{
assignmentBuilder.item(ma -> ma
.memberId(k)
.assignments(ta -> v.forEach(tp -> ta.item(i -> i
Expand All @@ -1101,12 +1112,19 @@ private void doMemberAssigment(
u.item(ud -> ud
.consumerId(at.consumerId)
.partitions(pt -> at.partitions.forEach(up ->
pt.item(pi -> pi.partitionId(up))))))))
))));
});
pt.item(pi -> pi.partitionId(up)))))))))))));

Array32FW<MemberAssignmentFW> assignment = assignmentBuilder.build();

if (KafkaConfiguration.DEBUG_CONSUMER)
{
assignment.forEach(c ->
{
System.out.printf("Assignment: MemberId - %s\n", c.memberId());
c.assignments().forEach(a -> a.partitions().forEach(System.out::println));
});
}

doConsumerInitialData(traceId, authorization, initialBud, assignment.sizeof(), 3,
assignment.buffer(), assignment.offset(), assignment.sizeof(), EMPTY_OCTETS);
}
Expand Down
17 changes: 17 additions & 0 deletions ...o/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaClientOffsetFetchFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,7 @@
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaBeginExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaDataExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaOffsetFetchBeginExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaOffsetFetchDataExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.KafkaResetExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.ProxyBeginExFW;
import io.aklivity.zilla.runtime.binding.kafka.internal.types.stream.ResetFW;
Expand Down Expand Up @@ -711,6 +712,8 @@ private int decodeOffsetFetchError(
client.decodeableResponseBytes -= error.sizeof();
assert client.decodeableResponseBytes >= 0;

client.onDecodeEmptyOffsetFetchResponse(traceId);

client.decoder = decodeOffsetFetchResponse;
client.nextResponseId++;
}
Expand Down Expand Up @@ -1729,6 +1732,20 @@ private void onDecodeOffsetFetchResponse(
}
}

private void onDecodeEmptyOffsetFetchResponse(
long traceId)
{
if (topicPartitions.isEmpty())
{
final KafkaDataExFW kafkaDataEx = kafkaDataExRW.wrap(extBuffer, 0, extBuffer.capacity())
.typeId(kafkaTypeId)
.offsetFetch(KafkaOffsetFetchDataExFW.Builder::build)
.build();

delegate.doApplicationData(traceId, authorization, kafkaDataEx);
}
}

public void onDecodeTopic(
long traceId)
{
Expand Down
34 changes: 32 additions & 2 deletions ...main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaMergedFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2010,6 +2010,8 @@ private void onTopicOffsetFetchDataChanged(
long traceId,
Array32FW<KafkaTopicPartitionOffsetFW> partitions)
{
offsetsByPartitionId.clear();

partitions.forEach(p -> offsetsByPartitionId.put(p.partitionId(),
new KafkaPartitionOffset(
topic,
Expand All @@ -2020,13 +2022,25 @@ private void onTopicOffsetFetchDataChanged(
p.metadata().asString())));

doFetchPartitionsIfNecessary(traceId);

fetchStreams.removeIf(f ->
{
boolean missing = !leadersByAssignedId.containsKey(f.partitionId);
if (missing)
{
f.doFetchInitialEndIfNecessary(traceId);
f.doFetchReplyResetIfNecessary(traceId);
}
return missing;
});
}

private void doFetchPartitionOffsets(
long traceId)
{
if (hasFetchCapability(capabilities))
{
offsetFetchStream.resetStreamIfNecessary(traceId);
offsetFetchStream.doOffsetFetchInitialBeginIfNecessary(traceId);
}
}
Expand Down Expand Up @@ -3190,7 +3204,7 @@ private KafkaUnmergedOffsetFetchStream(
private void doOffsetFetchInitialBeginIfNecessary(
long traceId)
{
if (!KafkaState.initialOpening(state))
if (!KafkaState.initialOpening(state) || KafkaState.closed(state))
{
doOffsetFetchInitialBegin(traceId);
}
Expand All @@ -3199,6 +3213,11 @@ private void doOffsetFetchInitialBeginIfNecessary(
private void doOffsetFetchInitialBegin(
long traceId)
{
if (KafkaState.closed(state))
{
state = 0;
}

assert state == 0;

state = KafkaState.openingInitial(state);
Expand Down Expand Up @@ -3332,7 +3351,8 @@ private void onOffsetFetchReplyData(
final Array32FW<KafkaTopicPartitionOffsetFW> partitions = kafkaOffsetFetchDataEx.partitions();
merged.onTopicOffsetFetchDataChanged(traceId, partitions);

doOffsetFetchReplyWindow(traceId, 0, replyMax);
doOffsetFetchInitialEndIfNecessary(traceId);
doOffsetFetchReplyResetIfNecessary(traceId);
}
}

Expand Down Expand Up @@ -3424,6 +3444,16 @@ private void doOffsetFetchReplyReset(
doReset(receiver, merged.routedId, merged.resolvedId, replyId, replySeq, replyAck, replyMax,
traceId, merged.authorization, EMPTY_EXTENSION);
}

private void resetStreamIfNecessary(
long traceId)
{
if (KafkaState.initialOpening(state))
{
doOffsetFetchInitialAbortIfNecessary(traceId);
doOffsetFetchReplyResetIfNecessary(traceId);
}
}
}

private final class KafkaUnmergedFetchStream
Expand Down
20 changes: 20 additions & 0 deletions .../src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheMergedIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -713,4 +713,24 @@ public void shouldAckMessageOffset() throws Exception
{
k3po.finish();
}

@Test
@Configuration("cache.options.merged.yaml")
@Specification({
"${app}/merged.fetch.unsubscribe/client",
"${app}/unmerged.group.fetch.unsubscribe/server"})
public void shouldUnsubscribeOnPartitionReassignment() throws Exception
{
k3po.finish();
}

@Test
@Configuration("cache.options.merged.yaml")
@Specification({
"${app}/merged.fetch.unsubscribe/client",
"${app}unmerged.group.fetch.assignment.incomplete/server"})
public void shouldCancelPreviousIncompleteOffsetFetchRequest() throws Exception
{
k3po.finish();
}
}
44 changes: 44 additions & 0 deletions .../zilla/specs/binding/kafka/streams/application/merged/merged.fetch.unsubscribe/client.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
#
# Copyright 2021-2023 Aklivity Inc.
#
# Aklivity 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.
#

connect "zilla://streams/app0"
option zilla:window 8192
option zilla:transmission "half-duplex"

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("test")
.groupId("client-1")
.consumerId("consumer-2")
.timeout(45000)
.partition(0, 1)
.partition(1, 1)
.partition(-1, 1)
.build()
.build()}

connected

read zilla:begin.ext ${kafka:matchBeginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("test")
.partition(0, 2, 2, 2, "test-meta")
.build()
.build()}
50 changes: 50 additions & 0 deletions .../zilla/specs/binding/kafka/streams/application/merged/merged.fetch.unsubscribe/server.rpt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
#
# Copyright 2021-2023 Aklivity Inc.
#
# Aklivity 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.
#

property deltaMillis 0L
property newTimestamp ${kafka:timestamp() + deltaMillis}

accept "zilla://streams/app0"
option zilla:window 8192
option zilla:transmission "half-duplex"

accepted

read zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("test")
.groupId("client-1")
.consumerId("consumer-2")
.timeout(45000)
.partition(0, 1)
.partition(1, 1)
.partition(-1, 1)
.build()
.build()}

connected

write zilla:begin.ext ${kafka:beginEx()
.typeId(zilla:id("kafka"))
.merged()
.capabilities("FETCH_ONLY")
.topic("test")
.partition(0, 2, 2, 2, "test-meta")
.build()
.build()}
write flush
Loading