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

Consumer group kafka function support #232

Merged
merged 4 commits into from
May 9, 2023
Merged

Consumer group kafka function support #232

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7107968
Checkpoint
akrambek May 5, 2023
6f8e5fb
Checkpoint
akrambek May 8, 2023
2e67dd0
Add unit tests
akrambek May 8, 2023
53ccb64
Update CacheFetchIT.java
jfallows May 9, 2023
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 ...a/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/CacheFetchIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -850,6 +850,7 @@ public void shouldReceiveMessagesWithIsolationReadUncommittedWhenAborted() throw
k3po.finish();
}

@Ignore("GitHub Actions")
@Test
@Configuration("cache.yaml")
@Specification({
Expand Down
219 changes: 218 additions & 1 deletion ...fka.spec/src/main/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,8 @@
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaFetchDataExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaFetchFlushExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaFlushExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaGroupBeginExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaGroupDataExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaMergedBeginExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaMergedDataExFW;
import io.aklivity.zilla.specs.binding.kafka.internal.types.stream.KafkaMergedFlushExFW;
Expand Down Expand Up @@ -607,6 +609,13 @@ public KafkaProduceBeginExBuilder produce()
return new KafkaProduceBeginExBuilder();
}

public KafkaGroupBeginExBuilder group()
{
beginExRW.kind(KafkaApi.GROUP.value());

return new KafkaGroupBeginExBuilder();
}

public byte[] build()
{
final KafkaBeginExFW beginEx = beginExRO;
Expand Down Expand Up @@ -964,6 +973,45 @@ private void ensureTransactionSet()
}
}
}

public final class KafkaGroupBeginExBuilder
{
private final KafkaGroupBeginExFW.Builder groupBeginExRW = new KafkaGroupBeginExFW.Builder();


private KafkaGroupBeginExBuilder()
{
groupBeginExRW.wrap(writeBuffer, KafkaBeginExFW.FIELD_OFFSET_PRODUCE, writeBuffer.capacity());
}

public KafkaGroupBeginExBuilder groupId(
String groupId)
{
groupBeginExRW.groupId(groupId);
return this;
}

public KafkaGroupBeginExBuilder protocol(
String protocol)
{
groupBeginExRW.protocol(protocol);
return this;
}

public KafkaGroupBeginExBuilder timeout(
int timeout)
{
groupBeginExRW.timeout(timeout);
return this;
}

public KafkaBeginExBuilder build()
{
final KafkaGroupBeginExFW groupBeginEx = groupBeginExRW.build();
beginExRO.wrap(writeBuffer, 0, groupBeginEx.limit());
return KafkaBeginExBuilder.this;
}
}
}

public static final class KafkaDataExBuilder
Expand Down Expand Up @@ -1021,14 +1069,20 @@ public KafkaProduceDataExBuilder produce()
return new KafkaProduceDataExBuilder();
}

public KafkaGroupDataExBuilder group()
{
dataExRW.kind(KafkaApi.GROUP.value());

return new KafkaGroupDataExBuilder();
}

public byte[] build()
{
final KafkaDataExFW dataEx = dataExRO;
final byte[] array = new byte[dataEx.sizeof()];
dataEx.buffer().getBytes(dataEx.offset(), array);
return array;
}

public final class KafkaFetchDataExBuilder
{
private final DirectBuffer keyRO = new UnsafeBuffer(0, 0);
Expand Down Expand Up @@ -1518,6 +1572,37 @@ public KafkaDataExBuilder build()
return KafkaDataExBuilder.this;
}
}

public final class KafkaGroupDataExBuilder
{
private final KafkaGroupDataExFW.Builder groupDataExRW = new KafkaGroupDataExFW.Builder();

private KafkaGroupDataExBuilder()
{
groupDataExRW.wrap(writeBuffer, KafkaDataExFW.FIELD_OFFSET_GROUP, writeBuffer.capacity());
}

public KafkaGroupDataExBuilder leaderId(
String leaderId)
{
groupDataExRW.leaderId(leaderId);
return this;
}

public KafkaGroupDataExBuilder memberId(
String memberId)
{
groupDataExRW.memberId(memberId);
return this;
}

public KafkaDataExBuilder build()
{
final KafkaGroupDataExFW groupDataEx = groupDataExRW.build();
dataExRO.wrap(writeBuffer, 0, groupDataEx.limit());
return KafkaDataExBuilder.this;
}
}
}

public static final class KafkaFlushExBuilder
Expand Down Expand Up @@ -1767,6 +1852,15 @@ public KafkaProduceDataExMatcherBuilder produce()
return matcherBuilder;
}

public KafkaGroupDataExMatchBuilder group()
{
final KafkaGroupDataExMatchBuilder matcherBuilder = new KafkaGroupDataExMatchBuilder();

this.kind = KafkaApi.GROUP.value();
this.caseMatcher = matcherBuilder::match;
return matcherBuilder;
}

public KafkaDataExMatcherBuilder typeId(
int typeId)
{
Expand Down Expand Up @@ -2507,6 +2601,55 @@ private boolean matchFilters(
return filters == null || filters == mergedDataEx.filters();
}
}

public final class KafkaGroupDataExMatchBuilder
{
private String16FW leaderId;
private String16FW memberId;

private KafkaGroupDataExMatchBuilder()
{
}

public KafkaGroupDataExMatchBuilder leaderId(
String leaderId)
{
this.leaderId = new String16FW(leaderId);
return this;
}

public KafkaGroupDataExMatchBuilder memberId(
String memberId)
{
this.memberId = new String16FW(memberId);
return this;
}

public KafkaDataExMatcherBuilder build()
{
return KafkaDataExMatcherBuilder.this;
}

private boolean match(
KafkaDataExFW dataEx)
{
final KafkaGroupDataExFW groupDataEx = dataEx.group();
return matchLeaderId(groupDataEx) &&
matchMemberId(groupDataEx);
}

private boolean matchLeaderId(
final KafkaGroupDataExFW groupDataEx)
{
return leaderId == null || leaderId.equals(groupDataEx.leaderId());
}

private boolean matchMemberId(
final KafkaGroupDataExFW groupDataEx)
{
return memberId == null || memberId.equals(groupDataEx.memberId());
}
}
}

public static final class KafkaFlushExMatcherBuilder
Expand Down Expand Up @@ -2781,6 +2924,15 @@ public KafkaProduceBeginExMatcherBuilder produce()
return matcherBuilder;
}

public KafkaGroupBeginExMatcherBuilder group()
{
final KafkaGroupBeginExMatcherBuilder matcherBuilder = new KafkaGroupBeginExMatcherBuilder();

this.kind = KafkaApi.GROUP.value();
this.caseMatcher = matcherBuilder::match;
return matcherBuilder;
}

public KafkaBeginExMatcherBuilder typeId(
int typeId)
{
Expand Down Expand Up @@ -3100,6 +3252,71 @@ private boolean matchPartition(

}

public final class KafkaGroupBeginExMatcherBuilder
{
private String16FW groupId;
private String16FW protocol;
private int timeout;

private KafkaGroupBeginExMatcherBuilder()
{
}

public KafkaGroupBeginExMatcherBuilder groupId(
String groupId)
{
this.groupId = new String16FW(groupId);
return this;
}

public KafkaGroupBeginExMatcherBuilder protocol(
String protocol)
{
this.protocol = new String16FW(protocol);
return this;
}

public KafkaGroupBeginExMatcherBuilder timeout(
int timeout)
{
this.timeout = timeout;
return this;
}

public KafkaBeginExMatcherBuilder build()
{
return KafkaBeginExMatcherBuilder.this;
}

private boolean match(
KafkaBeginExFW beginEx)
{
final KafkaGroupBeginExFW groupBeginEx = beginEx.group();
return matchGroupId(groupBeginEx) &&
matchGroupId(groupBeginEx) &&
matchProtocol(groupBeginEx) &&
matchTimeout(groupBeginEx);
}

private boolean matchGroupId(
final KafkaGroupBeginExFW groupBeginExFW)
{
return groupId == null || groupId.equals(groupBeginExFW.groupId());
}

private boolean matchProtocol(
final KafkaGroupBeginExFW groupBeginExFW)
{
return protocol == null || protocol.equals(groupBeginExFW.protocol());
}

private boolean matchTimeout(
final KafkaGroupBeginExFW groupBeginExFW)
{
return timeout == 0 || timeout == groupBeginExFW.timeout();
}
}

public final class KafkaMergedBeginExMatcherBuilder
{
private KafkaCapabilities capabilities;
Expand Down
20 changes: 18 additions & 2 deletions specs/binding-kafka.spec/src/main/resources/META-INF/zilla/kafka.idl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,7 +171,8 @@ scope kafka
META (3),
DESCRIBE (32),
FETCH (1),
PRODUCE (0)
PRODUCE (0),
GROUP (10)
}

union KafkaBeginEx switch (uint8) extends core::stream::Extension
Expand All @@ -182,6 +183,7 @@ scope kafka
case 32: kafka::stream::KafkaDescribeBeginEx describe;
case 1: kafka::stream::KafkaFetchBeginEx fetch;
case 0: kafka::stream::KafkaProduceBeginEx produce;
case 10: kafka::stream::KafkaGroupBeginEx group;
}

union KafkaDataEx switch (uint8) extends core::stream::Extension
Expand All @@ -191,6 +193,7 @@ scope kafka
case 32: kafka::stream::KafkaDescribeDataEx describe;
case 1: kafka::stream::KafkaFetchDataEx fetch;
case 0: kafka::stream::KafkaProduceDataEx produce;
case 10: kafka::stream::KafkaGroupDataEx group;
}

union KafkaFlushEx switch (uint8) extends core::stream::Extension
Expand Down Expand Up @@ -278,7 +281,7 @@ scope kafka
int64 timestamp = 0; // INIT only
int32 headersSizeMax = 4; // INIT only
int64 producerId = -1; // INIT only
int64 filters = -1; // INIT only
int64 filters = -1; // INIT only
KafkaOffset partition; // INIT only
KafkaKey key; // INIT only
KafkaDelta delta; // INIT + FIN
Expand Down Expand Up @@ -315,5 +318,18 @@ scope kafka
KafkaKey key;
KafkaHeader[] headers;
}

struct KafkaGroupBeginEx
{
string16 groupId;
string16 protocol;
int32 timeout;
}

struct KafkaGroupDataEx
{
string16 leaderId;
string16 memberId;
}
}
}
Loading