Update Kafka Client to 2.8.0 #1588

eolivelli · 2022-11-23T15:17:49Z

Summary of changes:

Bump client dependency from 2.1.x to 2.8.x
The Java Model for all the Request/Response changed to a Fluent/Builder style, this requires many changes
NotLeaderForPartitionException -> NotLeaderOrFollowerException
Copy new version of MockTime from Kafka codebase
ListOffsetRequest -> ListOffsetsRequest, ListOffsetResponse -> ListOffsetsResponse
LeaveGroup -> now support multiple members

github-actions · 2022-11-23T15:18:07Z

@eolivelli:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java

...java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/CorrelationIdGenerator.java

Demogorgon314 · 2022-11-30T10:09:19Z

Overall, LGTM. I will wait for the unit's test to pass.

eolivelli · 2022-12-01T10:34:11Z

@Demogorgon314 @BewareMyPower this patch is finally ready for review.
All the tests are passing locally on my laptop and Codacy is happy

codecov · 2022-12-01T11:23:16Z

Codecov Report

Merging #1588 (64ce7c8) into master (b4dbd6d) will decrease coverage by 0.35%.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master    #1588      +/-   ##
============================================
- Coverage     15.34%   14.98%   -0.36%     
  Complexity      592      592              
============================================
  Files           164      164              
  Lines         11913    12199     +286     
  Branches       1102     1112      +10     
============================================
  Hits           1828     1828              
- Misses         9931    10217     +286     
  Partials        154      154

Impacted Files	Coverage Δ
...streamnative/pulsar/handlers/kop/AdminManager.java	`0.40% <0.00%> (-0.01%)`	⬇️
...ative/pulsar/handlers/kop/KafkaCommandDecoder.java	`0.34% <0.00%> (+<0.01%)`	⬆️
...ative/pulsar/handlers/kop/KafkaRequestHandler.java	`1.09% <0.00%> (-0.07%)`	⬇️
...ive/pulsar/handlers/kop/MessagePublishContext.java	`0.00% <0.00%> (ø)`
...ndlers/kop/coordinator/group/GroupCoordinator.java	`0.00% <0.00%> (ø)`
.../handlers/kop/coordinator/group/GroupMetadata.java	`51.78% <0.00%> (-0.19%)`	⬇️
...andlers/kop/coordinator/group/JoinGroupResult.java	`0.00% <0.00%> (ø)`
...rs/kop/coordinator/transaction/PendingRequest.java	`0.00% <0.00%> (ø)`
...s/kop/coordinator/transaction/ResponseContext.java	`0.00% <0.00%> (ø)`
...r/transaction/TransactionMarkerChannelHandler.java	`0.00% <0.00%> (ø)`
... and 9 more

eolivelli · 2022-12-01T12:02:41Z

CI passed

BewareMyPower · 2022-12-01T12:30:16Z

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

+        Set<String> members = new HashSet<>();
+        if (!data.memberId().isEmpty()) {
+            // old clients
+            members.add(data.memberId());
+        }
+        data.members().forEach(memberIdentity -> {
+            members.add(memberIdentity.memberId());
+        });


We can use functional style here

Set<String> members = data.members().stream() .map(LeaveGroupRequestData.MemberIdentity::memberId) .collect(Collectors.toSet()); if (!data.memberId().isEmpty()) { // old clients members.add(data.memberId()); }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

BewareMyPower · 2022-12-01T13:20:22Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

+                        .filter(p -> p.partitionIndex() == topicPartition1.partition())
+                        .findFirst()
+                        .get()
+                        .errorCode(),
            Errors.UNKNOWN_TOPIC_OR_PARTITION);


Suggested change

Errors.UNKNOWN_TOPIC_OR_PARTITION);

Errors.UNKNOWN_TOPIC_OR_PARTITION.code());

BewareMyPower · 2022-12-01T13:20:49Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

+                                .filter(p -> p.partitionIndex() == topicPartition2.partition())
+                                .findFirst()
+                                .get()
+                                .errorCode(),
            Errors.UNKNOWN_TOPIC_OR_PARTITION);


Suggested change

Errors.UNKNOWN_TOPIC_OR_PARTITION);

Errors.UNKNOWN_TOPIC_OR_PARTITION.code());

BewareMyPower · 2022-12-01T13:21:09Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

            Errors.UNKNOWN_TOPIC_OR_PARTITION);

        // invalid partition id 1.
        CompletableFuture<AbstractResponse> invalidResponse2 = new CompletableFuture<>();
        checkInvalidPartition(invalidResponse2, topicName, 1);
        TopicPartition topicPartition2 = new TopicPartition(topicName, 1);
        AbstractResponse response2 = invalidResponse2.get();
-        assertEquals(((OffsetCommitResponse) response2).responseData().get(topicPartition2),
+                assertEquals(((OffsetCommitResponse) response2)


Suggested change

assertEquals(((OffsetCommitResponse) response2)

assertEquals(((OffsetCommitResponse) response2)

Adjust indent.

BewareMyPower · 2022-12-01T13:23:04Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaAuthorizationTestBase.java

@@ -680,6 +680,7 @@ public void testDescribeConfigFailed() throws PulsarAdminException {
        try {
            describeConfigsResult.all().get();
        } catch (Exception ex) {
+            log.error("Error", ex);


Suggested change

log.error("Error", ex);

Remove this line because the exception is expected. Or just use log.info("Error: {}", ex.getMessage(). BTW, I found here we should add a fail() in the try block.

@BewareMyPower I didn't add "fail()" for this PR, because with "fail()" actually the test will fail.
We can address this problem in the test in a follow up work.
it is not strictly related to this patch

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java

BewareMyPower · 2022-12-01T13:29:37Z

My review is done, PTAL.

eolivelli

Thank you @BewareMyPower for you quick feedback
I have addressed all of your comments.

eolivelli · 2022-12-01T14:13:07Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaAuthorizationTestBase.java

@@ -680,6 +680,7 @@ public void testDescribeConfigFailed() throws PulsarAdminException {
        try {
            describeConfigsResult.all().get();
        } catch (Exception ex) {
+            log.error("Error", ex);


tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaListenerNameTest.java

eolivelli · 2022-12-01T14:16:10Z

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

+        Set<String> members = new HashSet<>();
+        if (!data.memberId().isEmpty()) {
+            // old clients
+            members.add(data.memberId());
+        }
+        data.members().forEach(memberIdentity -> {
+            members.add(memberIdentity.memberId());
+        });


eolivelli · 2022-12-01T14:42:59Z

@BewareMyPower what about the "snyk" failures ? before my last commit it was passing

eolivelli · 2022-12-01T14:43:06Z

@BewareMyPower what about the "snyk" failures ? before my last commit it was passing

Demogorgon314

LGTM. Just left some small comments.

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

...pl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupCoordinator.java

Demogorgon314 · 2022-12-01T15:20:28Z

...pl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupCoordinator.java

    ) {
        return validateGroupStatus(groupId, ApiKeys.LEAVE_GROUP).map(CompletableFuture::completedFuture
        ).orElseGet(() -> groupManager.getGroup(groupId).map(group -> group.inLock(() -> {
-            if (group.is(Dead) || !group.has(memberId)) {
+            if (group.is(Dead)) {
+                return CompletableFuture.completedFuture(Errors.UNKNOWN_MEMBER_ID);


The error should be Errors.COORDINATOR_NOT_AVAILABLE?

previously it was UNKNOWN_MEMBER_ID, I didn't want to change the behaviour

Sure, I just found the Kafka 2.8 code is using Errors.COORDINATOR_NOT_AVAILABLE. We can change it in next PR if necessary.

https://github.com/apache/kafka/blob/2.8/core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala#L470-L471

let me follow up with another patch on this

@Demogorgon314 finally I updated this value in this PR

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaResponseUtils.java

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

eolivelli

I have addresses your comments.
I hope I haven't missed any of them

thank you @Demogorgon314 @BewareMyPower

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

eolivelli · 2022-12-01T15:50:51Z

...pl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupCoordinator.java

    ) {
        return validateGroupStatus(groupId, ApiKeys.LEAVE_GROUP).map(CompletableFuture::completedFuture
        ).orElseGet(() -> groupManager.getGroup(groupId).map(group -> group.inLock(() -> {
-            if (group.is(Dead) || !group.has(memberId)) {
+            if (group.is(Dead)) {
+                return CompletableFuture.completedFuture(Errors.UNKNOWN_MEMBER_ID);


previously it was UNKNOWN_MEMBER_ID, I didn't want to change the behaviour

...pl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupCoordinator.java

eolivelli · 2022-12-01T15:53:59Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

+                        .filter(p -> p.partitionIndex() == topicPartition1.partition())
+                        .findFirst()
+                        .get()
+                        .errorCode(),
            Errors.UNKNOWN_TOPIC_OR_PARTITION);


eolivelli · 2022-12-01T15:54:25Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

            Errors.UNKNOWN_TOPIC_OR_PARTITION);

        // invalid partition id 1.
        CompletableFuture<AbstractResponse> invalidResponse2 = new CompletableFuture<>();
        checkInvalidPartition(invalidResponse2, topicName, 1);
        TopicPartition topicPartition2 = new TopicPartition(topicName, 1);
        AbstractResponse response2 = invalidResponse2.get();
-        assertEquals(((OffsetCommitResponse) response2).responseData().get(topicPartition2),
+                assertEquals(((OffsetCommitResponse) response2)


eolivelli · 2022-12-01T15:54:43Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

+                                .filter(p -> p.partitionIndex() == topicPartition2.partition())
+                                .findFirst()
+                                .get()
+                                .errorCode(),
            Errors.UNKNOWN_TOPIC_OR_PARTITION);


tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java

eolivelli · 2022-12-01T16:02:21Z

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaAuthorizationTestBase.java

@@ -680,6 +680,7 @@ public void testDescribeConfigFailed() throws PulsarAdminException {
        try {
            describeConfigsResult.all().get();
        } catch (Exception ex) {
+            log.error("Error", ex);


@BewareMyPower I didn't add "fail()" for this PR, because with "fail()" actually the test will fail.
We can address this problem in the test in a follow up work.
it is not strictly related to this patch

eolivelli · 2022-12-01T17:13:36Z

The tests that failed on CI pass on my laptop

eolivelli · 2022-12-02T09:56:46Z

@BewareMyPower is the failed test a flaky test ?
it doesn't fail locally for me

This reverts commit 47d4995.

Co-authored-by: Yunze Xu <xyzinfernity@163.com>

eolivelli · 2022-12-02T12:43:34Z

I have rebased on master

Demogorgon314 · 2022-12-02T13:46:46Z

@eolivelli I can reproduce this failed test in my locally.

eolivelli · 2022-12-02T15:26:31Z

@eolivelli I can reproduce this failed test in my locally.

I will run the test many times. thanks for your feedback.
The test is pretty straight forward

eolivelli · 2022-12-02T19:08:06Z

@Demogorgon314 I have fixed the test.
it seems that the 2.8.0 client eagerly pre-fetches the messages and so it is able to pre-fetch the records before we delete the partition.

eolivelli · 2022-12-02T20:38:37Z

@BewareMyPower @Demogorgon314 can you please restart CI ?
I believe that this time we hit a flaky test.

eolivelli · 2022-12-02T20:41:40Z

the same test did not fail here
#1606

eolivelli · 2022-12-04T07:57:13Z

All tests passed.

Summary of changes: - Bump client dependency from 2.1.x to 2.8.x - The Java Model for all the Request/Response changed to a Fluent/Builder style, this requires many changes - NotLeaderForPartitionException -> NotLeaderOrFollowerException - Copy new version of MockTime from Kafka codebase - ListOffsetRequest -> ListOffsetsRequest, ListOffsetResponse -> ListOffsetsResponse - LeaveGroup -> now support multiple members (cherry picked from commit 66efad6)

github-actions bot assigned eolivelli Nov 23, 2022

github-actions bot added the doc-info-missing This pr needs to mark a document option in description label Nov 23, 2022

eolivelli force-pushed the impl/upgrade-kafka-dep-280 branch from 436e36c to 109f697 Compare November 29, 2022 08:38

Demogorgon314 reviewed Nov 30, 2022

View reviewed changes

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java Outdated Show resolved Hide resolved

Demogorgon314 reviewed Nov 30, 2022

View reviewed changes

...java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/CorrelationIdGenerator.java Outdated Show resolved Hide resolved

eolivelli mentioned this pull request Nov 30, 2022

Support transaction producer fence #913

Closed

2 tasks

eolivelli marked this pull request as ready for review December 1, 2022 10:33

eolivelli requested a review from BewareMyPower as a code owner December 1, 2022 10:33

eolivelli requested a review from Demogorgon314 December 1, 2022 10:33

BewareMyPower changed the title ~~Update Kafka Client to 2.8.0 - PREVIEW~~ Update Kafka Client to 2.8.0 Dec 1, 2022

BewareMyPower reviewed Dec 1, 2022

View reviewed changes

eolivelli commented Dec 1, 2022

View reviewed changes

Demogorgon314 reviewed Dec 1, 2022

View reviewed changes

eolivelli commented Dec 1, 2022

View reviewed changes

Demogorgon314 previously approved these changes Dec 2, 2022

View reviewed changes

Demogorgon314 added no-need-doc This pr does not need any document release/2.11 and removed doc-info-missing This pr needs to mark a document option in description labels Dec 2, 2022

BewareMyPower previously approved these changes Dec 2, 2022

View reviewed changes

eolivelli and others added 12 commits December 2, 2022 13:42

Revert "Make KafkaAuthorizationTestBase less flaky"

217b218

This reverts commit 47d4995.

checkstyle

de82f60

more tests passing

2ae24b4

fix test

9f426ba

Fix some tests

a0dd649

Fix Codacity

3436db1

fix test

48267fb

Apply suggestions from code review

8e0d9e2

Co-authored-by: Yunze Xu <xyzinfernity@163.com>

Code review feedback

92546ec

fix build

78e4b42

CR feedback

9b2b368

Update GroupCoordinator

214bae8

eolivelli force-pushed the impl/upgrade-kafka-dep-280 branch from de820da to 214bae8 Compare December 2, 2022 12:43

eolivelli added 2 commits December 2, 2022 18:14

Reduce flakyness

c4c181c

checkstyle

64ce7c8

Demogorgon314 approved these changes Dec 4, 2022

View reviewed changes

Demogorgon314 merged commit 66efad6 into streamnative:master Dec 4, 2022

eolivelli deleted the impl/upgrade-kafka-dep-280 branch December 4, 2022 14:30

Demogorgon314 mentioned this pull request Dec 20, 2022

[BUG] log-end-offset becomes a negative number #1634

Closed

Demogorgon314 added the cherry-picked/branch-2.11 label Feb 6, 2023

BewareMyPower mentioned this pull request Mar 2, 2023

[BUG] KStreams doesn't work with KoP #1743

Closed

	Errors.UNKNOWN_TOPIC_OR_PARTITION);
	Errors.UNKNOWN_TOPIC_OR_PARTITION.code());

	assertEquals(((OffsetCommitResponse) response2)
	assertEquals(((OffsetCommitResponse) response2)

Update Kafka Client to 2.8.0 #1588

Update Kafka Client to 2.8.0 #1588

Conversation

eolivelli commented Nov 23, 2022 • edited Loading

github-actions bot commented Nov 23, 2022

Demogorgon314 commented Nov 30, 2022

eolivelli commented Dec 1, 2022 • edited Loading

codecov bot commented Dec 1, 2022 • edited Loading

Codecov Report

eolivelli commented Dec 1, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

BewareMyPower commented Dec 1, 2022

eolivelli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli commented Dec 1, 2022

eolivelli commented Dec 1, 2022

Demogorgon314 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eolivelli commented Dec 1, 2022

eolivelli commented Dec 2, 2022

eolivelli commented Dec 2, 2022

Demogorgon314 commented Dec 2, 2022

eolivelli commented Dec 2, 2022

eolivelli commented Dec 2, 2022

eolivelli commented Dec 2, 2022

eolivelli commented Dec 2, 2022

eolivelli commented Dec 4, 2022

eolivelli commented Nov 23, 2022 •

edited

Loading

eolivelli commented Dec 1, 2022 •

edited

Loading

codecov bot commented Dec 1, 2022 •

edited

Loading