Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709 #1981

gaoran10 · 2023-07-27T13:04:40Z

(cherry picked from commit 71b77b2)

Motivation

Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709.

Modifications

Describe the modifications you've done.

Documentation

Check the box below.

Need to update docs?

doc-required

(If you need help on updating docs, create a doc issue)
no-need-doc

(Please explain why)
doc

(If this PR contains doc changes)

codecov · 2023-07-27T13:13:00Z

Codecov Report

Merging #1981 (c0caa7e) into master (3b22e79) will decrease coverage by 0.19%.
Report is 2 commits behind head on master.
The diff coverage is 0.00%.

@@             Coverage Diff              @@
##             master    #1981      +/-   ##
============================================
- Coverage     17.26%   17.07%   -0.19%     
- Complexity      726      728       +2     
============================================
  Files           190      191       +1     
  Lines         14036    14218     +182     
  Branches       1318     1329      +11     
============================================
+ Hits           2423     2428       +5     
- Misses        11437    11614     +177     
  Partials        176      176

Files Changed	Coverage Δ
...streamnative/pulsar/handlers/kop/DelayedFetch.java	`0.00% <ø> (ø)`
...ative/pulsar/handlers/kop/KafkaCommandDecoder.java	`0.30% <0.00%> (-0.01%)`	⬇️
...ative/pulsar/handlers/kop/KafkaRequestHandler.java	`1.00% <0.00%> (-0.06%)`	⬇️
...rs/kop/exceptions/KoPTopicInitializeException.java	`0.00% <0.00%> (ø)`
...tive/pulsar/handlers/kop/storage/PartitionLog.java	`8.35% <0.00%> (+0.41%)`	⬆️
...sar/handlers/kop/storage/ProducerStateManager.java	`0.00% <0.00%> (ø)`
...ve/pulsar/handlers/kop/storage/ReplicaManager.java	`0.00% <0.00%> (ø)`
.../pulsar/handlers/kop/utils/KafkaResponseUtils.java	`0.00% <0.00%> (ø)`
...ulsar/handlers/kop/utils/MessageMetadataUtils.java	`8.00% <0.00%> (-0.83%)`	⬇️

(cherry picked from commit 71b77b2)

Demogorgon314 · 2023-08-03T09:41:14Z

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

+ .setPort(result.node.port())
+ .setErrorCode(result.error.code())
+ .setErrorMessage(result.error.message())
+ .setKey(coordinatorKey));
 });
 } else if (request.data().keyType() == FindCoordinatorRequest.CoordinatorType.GROUP.id()) {
 authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.GROUP, request.data().key()))


Suggested change

authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.GROUP, request.data().key()))

authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.GROUP, coordinatorKey))

This change should be able to fix the group authorize exception

Demogorgon314 · 2023-08-03T09:46:07Z

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

@@ -311,6 +312,7 @@ protected final void internalSetup(boolean startBroker) throws Exception {
 createClient();

 MetadataUtils.createOffsetMetadataIfMissing(conf.getKafkaMetadataTenant(), admin, clusterData, this.conf);
+


Unused change

Demogorgon314 · 2023-08-03T09:47:57Z

pom.xml

@@ -48,7 +48,7 @@
 <!-- dependencies -->
 <jackson.version>2.14.2</jackson.version>
 <jackson-databind.version>2.14.2</jackson-databind.version>
- <kafka.version>2.8.0</kafka.version>
+ <kafka.version>3.4.0</kafka.version>


Should we upgrade to Kafka 3.4.1?

Demogorgon314 · 2023-08-03T09:49:45Z

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

@@ -1187,7 +1198,8 @@ public CompletableFuture<Long> forcePurgeAbortTx() {
 public CompletableFuture<Long> recoverTxEntries(
 long offset,
 Executor executor) {
- if (!kafkaConfig.isKafkaTransactionCoordinatorEnabled()) {
+ if (!kafkaConfig.isKafkaTransactionCoordinatorEnabled()


Why we need the isInterceptorConfigured check?

I'm not sure about this change, there are two implementations MLTransactionSequenceIdGenerator and ManagedLedgerInterceptorImpl, maybe need more context from Enrico.

if you don't have the interceptor then you don't have the "index" (Kafka offset) in the records, and it is not possible to perform recovery.
IIRC this was because there are some tests about having legacy/broken configuration.

Now that KOP/S4K are mature it is not possible that people don't configure the interceptor,
maybe you can drop the check (but I bet that some tests will fail)

Thanks, I'll drop it and fix the tests.

Demogorgon314 · 2023-08-03T09:57:57Z

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

@@ -184,22 +188,52 @@ public static DescribeGroupsResponse newDescribeGroups(
 return new DescribeGroupsResponse(data);
 }

- public static FindCoordinatorResponse newFindCoordinator(Node node) {
+ public static FindCoordinatorResponse newFindCoordinator(List<String> coordinatorKeys,


Can we reuse this method? It looks like this method is not used anymore.

It seems that this method is used for Pulsar proxy, I remove this method and create a new one for generating Coordinator results.

if the method is not used then we can drop it.

eolivelli

LGTM

awesome

gaoran10 · 2023-08-04T13:28:06Z

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

- return CompletableFuture.completedFuture(Long.valueOf(0));
+ return CompletableFuture.completedFuture(0L);
+ }
+ if (!isBrokerIndexMetadataInterceptorConfigured(persistentTopic.getBrokerService())) {


If we don't have this check, the UpgradeTest will encounter an NPE problem while getting the current offset in method fetchOldestAvailableIndexFromTopic, but we can't disable the Kafka transaction in that test right now, because newly Kafka client requires the producer id.

Actually, I think this check is necessary because the Kafka transaction can't work without continuously offset, maybe we can add this check while starting the broker. @Demogorgon314

…streamnative#1981) (cherry picked from commit 71b77b2) ### Motivation Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709. --------- Co-authored-by: Enrico Olivelli <enrico.olivelli@datastax.com> (cherry picked from commit 19801c1)

gaoran10 requested review from BewareMyPower and Demogorgon314 as code owners July 27, 2023 13:04

gaoran10 self-assigned this Jul 27, 2023

gaoran10 added release/2.11 release/2.10.4 release/3.0 labels Jul 27, 2023

github-actions bot added the no-need-doc This pr does not need any document label Jul 27, 2023

gaoran10 force-pushed the update_kafka_wireprotocol_and_impl_kip_699_and_709 branch from 3a4e749 to 930e502 Compare July 30, 2023 05:20

gaoran10 mentioned this pull request Jul 30, 2023

[transactions] Implement KIP-664 listTransactions #1984

Open

3 tasks

gaoran10 force-pushed the update_kafka_wireprotocol_and_impl_kip_699_and_709 branch from fce6f45 to 325f515 Compare August 1, 2023 01:53

eolivelli and others added 2 commits August 3, 2023 15:24

Update Kafka wireprotocol to 3.4.0 and implement KIP-699 and KIP-709

7b901c6

(cherry picked from commit 71b77b2)

add ignore annotation for test testAlterClientQuotas

97b819f

gaoran10 force-pushed the update_kafka_wireprotocol_and_impl_kip_699_and_709 branch from 325f515 to 97b819f Compare August 3, 2023 07:25

fix test

161ed32

Demogorgon314 reviewed Aug 3, 2023

View reviewed changes

fix

209fdc4

eolivelli approved these changes Aug 4, 2023

View reviewed changes

gaoran10 added 2 commits August 4, 2023 19:20

remove ManagedLedgerInterceptor check

b3e990c

add a new check for AppendIndexMetadataInterceptor

c0caa7e

gaoran10 commented Aug 4, 2023

View reviewed changes

Demogorgon314 approved these changes Aug 7, 2023

View reviewed changes

gaoran10 merged commit 19801c1 into streamnative:master Aug 7, 2023
17 of 19 checks passed

gaoran10 deleted the update_kafka_wireprotocol_and_impl_kip_699_and_709 branch August 7, 2023 02:11

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709 #1981

Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709 #1981

gaoran10 commented Jul 27, 2023

codecov bot commented Jul 27, 2023 •

edited

Loading

Demogorgon314 Aug 3, 2023

gaoran10 Aug 4, 2023

Demogorgon314 Aug 3, 2023

gaoran10 Aug 4, 2023

Demogorgon314 Aug 3, 2023

gaoran10 Aug 4, 2023

Demogorgon314 Aug 3, 2023

gaoran10 Aug 4, 2023 •

edited

Loading

eolivelli Aug 4, 2023

gaoran10 Aug 4, 2023 •

edited

Loading

Demogorgon314 Aug 3, 2023

gaoran10 Aug 4, 2023 •

edited

Loading

eolivelli Aug 4, 2023

eolivelli left a comment

gaoran10 Aug 4, 2023 •

edited

Loading

	authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.GROUP, request.data().key()))
	authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.GROUP, coordinatorKey))

		@@ -311,6 +312,7 @@ protected final void internalSetup(boolean startBroker) throws Exception {
		createClient();

		MetadataUtils.createOffsetMetadataIfMissing(conf.getKafkaMetadataTenant(), admin, clusterData, this.conf);

Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709 #1981

Update Kafka wire protocol to 3.4.0 and implement KIP-699 and KIP-709 #1981

Conversation

gaoran10 commented Jul 27, 2023

Motivation

Modifications

Documentation

codecov bot commented Jul 27, 2023 • edited Loading

Codecov Report

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

gaoran10 Aug 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaoran10 Aug 4, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gaoran10 Aug 4, 2023 • edited Loading

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

gaoran10 Aug 4, 2023 • edited Loading

Choose a reason for hiding this comment

codecov bot commented Jul 27, 2023 •

edited

Loading

gaoran10 Aug 4, 2023 •

edited

Loading

gaoran10 Aug 4, 2023 •

edited

Loading

gaoran10 Aug 4, 2023 •

edited

Loading

gaoran10 Aug 4, 2023 •

edited

Loading