You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "ijuma (via GitHub)" <gi...@apache.org> on 2025/01/12 21:13:26 UTC
[PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
ijuma opened a new pull request, #18497:
URL: https://github.com/apache/kafka/pull/18497
This is not complete yet, I want to see the results of a full test run. Aside from that,
I also need to add new generator tests.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922745759
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -35,7 +35,6 @@ import scala.jdk.CollectionConverters._
import scala.collection._
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success, Try}
-import org.apache.kafka.common.requests.{AbstractControlRequest, LeaderAndIsrRequest}
Review Comment:
It seems we can remove `AbstractControlRequest` in this PR.
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -6265,45 +6005,10 @@ class ReplicaManagerTest {
assertEquals(expectedTopicId, fetchState.get.topicId)
}
- @ParameterizedTest
- @ValueSource(booleans = Array(true, false))
- def testPartitionFetchStateUpdatesWithTopicIdChanges(startsWithTopicId: Boolean): Unit = {
- val aliveBrokersIds = Seq(0, 1)
- val replicaManager = setupReplicaManagerWithMockedPurgatories(new MockTimer(time),
- brokerId = 0, aliveBrokersIds)
- try {
- val tp = new TopicPartition(topic, 0)
- val leaderAndIsr = new LeaderAndIsr(1, aliveBrokersIds.toList.map(Int.box).asJava)
-
- // This test either starts with a topic ID in the PartitionFetchState and removes it on the next request (startsWithTopicId)
- // or does not start with a topic ID in the PartitionFetchState and adds one on the next request (!startsWithTopicId)
- val startingId = if (startsWithTopicId) topicId else Uuid.ZERO_UUID
- val startingIdOpt = if (startsWithTopicId) Some(topicId) else None
- val leaderAndIsrRequest1 = makeLeaderAndIsrRequest(startingId, tp, aliveBrokersIds, leaderAndIsr)
- val leaderAndIsrResponse1 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest1, (_, _) => ())
- assertEquals(Errors.NONE, leaderAndIsrResponse1.error)
-
- assertFetcherHasTopicId(replicaManager.replicaFetcherManager, tp, startingIdOpt)
-
- val endingId = if (!startsWithTopicId) topicId else Uuid.ZERO_UUID
- val endingIdOpt = if (!startsWithTopicId) Some(topicId) else None
- val leaderAndIsrRequest2 = makeLeaderAndIsrRequest(endingId, tp, aliveBrokersIds, leaderAndIsr)
- val leaderAndIsrResponse2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest2, (_, _) => ())
- assertEquals(Errors.NONE, leaderAndIsrResponse2.error)
-
- assertFetcherHasTopicId(replicaManager.replicaFetcherManager, tp, endingIdOpt)
- } finally {
- replicaManager.shutdown(checkpointHW = false)
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = Array(true, false))
- def testReplicaAlterLogDirsWithAndWithoutIds(usesTopicIds: Boolean): Unit = {
+ @Test
+ def testReplicaAlterLogDirs(): Unit = {
val tp = new TopicPartition(topic, 0)
- val version = if (usesTopicIds) LeaderAndIsrRequestData.HIGHEST_SUPPORTED_VERSION else 4.toShort
- val topicId = if (usesTopicIds) this.topicId else Uuid.ZERO_UUID
- val topicIdOpt = if (usesTopicIds) Some(topicId) else None
+ val topicId = this.topicId
Review Comment:
This is unnecessary
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -144,10 +151,12 @@ public static RequestHeader parse(ByteBuffer buffer) {
header.size = Math.max(buffer.position() - bufferStartPositionForHeader, 0);
return header;
} catch (UnsupportedVersionException e) {
- throw new InvalidRequestException("Unknown API key " + apiKey, e);
+ throw new InvalidRequestException("Unknown API key " + apiKeyId, e);
} catch (Throwable ex) {
- throw new InvalidRequestException("Error parsing request header. Our best guess of the apiKey is: " +
- apiKey, ex);
+ if (ex instanceof InvalidRequestException)
Review Comment:
How about adding `InvalidRequestException` to the catch block?
```java
} catch (UnsupportedVersionException e) {
throw new InvalidRequestException("Unknown API key " + apiKeyId, e);
} catch (InvalidRequestException e) {
throw e;
} catch (Throwable ex) {
throw new InvalidRequestException("Error parsing request header. Our best guess of the apiKeyId is: " +
apiKeyId, ex);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18497:
URL: https://github.com/apache/kafka/pull/18497#issuecomment-2601879284
Test failures are unrelated - kicked off the tests again to try for a green build.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922576857
##########
clients/src/main/resources/common/message/LeaderAndIsrRequest.json:
##########
@@ -16,84 +16,6 @@
{
"apiKey": 4,
"type": "request",
- "listeners": ["zkBroker"],
"name": "LeaderAndIsrRequest",
- // Version 1 adds IsNew.
- //
- // Version 2 adds broker epoch and reorganizes the partitions by topic.
- //
- // Version 3 adds AddingReplicas and RemovingReplicas.
- //
- // Version 4 is the first flexible version.
- //
- // Version 5 adds Topic ID and Type to the TopicStates, as described in KIP-516.
- //
- // Version 6 adds LeaderRecoveryState as described in KIP-704.
- //
- // Version 7 adds KRaft Controller ID field as part of KIP-866
- "validVersions": "0-7",
- "flexibleVersions": "4+",
- "fields": [
- { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The current controller ID." },
- { "name": "isKRaftController", "type": "bool", "versions": "7+", "default": "false",
- "about": "If KRaft controller id is used during migration. See KIP-866." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The current controller epoch." },
- { "name": "BrokerEpoch", "type": "int64", "versions": "2+", "ignorable": true, "default": "-1",
- "about": "The current broker epoch." },
- { "name": "Type", "type": "int8", "versions": "5+",
- "about": "The type that indicates whether all topics are included in the request."},
- { "name": "UngroupedPartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "0-1",
- "about": "The state of each partition, in a v0 or v1 message." },
- // In v0 or v1 requests, each partition is listed alongside its topic name.
- // In v2+ requests, partitions are organized by topic, so that each topic name
- // only needs to be listed once.
- { "name": "TopicStates", "type": "[]LeaderAndIsrTopicState", "versions": "2+",
- "about": "Each topic.", "fields": [
- { "name": "TopicName", "type": "string", "versions": "2+", "entityType": "topicName",
- "about": "The topic name." },
- { "name": "TopicId", "type": "uuid", "versions": "5+", "ignorable": true,
- "about": "The unique topic ID." },
- { "name": "PartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "2+",
- "about": "The state of each partition." }
- ]},
- { "name": "LiveLeaders", "type": "[]LeaderAndIsrLiveLeader", "versions": "0+",
- "about": "The current live leaders.", "fields": [
- { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The leader's broker ID." },
- { "name": "HostName", "type": "string", "versions": "0+",
- "about": "The leader's hostname." },
- { "name": "Port", "type": "int32", "versions": "0+",
- "about": "The leader's port." }
- ]}
- ],
- "commonStructs": [
- { "name": "LeaderAndIsrPartitionState", "versions": "0+", "fields": [
- { "name": "TopicName", "type": "string", "versions": "0-1", "entityType": "topicName", "ignorable": true,
- "about": "The topic name. This is only present in v0 or v1." },
- { "name": "PartitionIndex", "type": "int32", "versions": "0+",
- "about": "The partition index." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The controller epoch." },
- { "name": "Leader", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The broker ID of the leader." },
- { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
- "about": "The leader epoch." },
- { "name": "Isr", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The in-sync replica IDs." },
- { "name": "PartitionEpoch", "type": "int32", "versions": "0+",
- "about": "The current epoch for the partition. The epoch is a monotonically increasing value which is incremented after every partition change. (Since the LeaderAndIsr request is only used by the legacy controller, this corresponds to the zkVersion)." },
- { "name": "Replicas", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The replica IDs." },
- { "name": "AddingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are adding this partition to, or null if no replicas are being added." },
- { "name": "RemovingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are removing this partition from, or null if no replicas are being removed." },
- { "name": "IsNew", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
- "about": "Whether the replica should have existed on the broker or not." },
- { "name": "LeaderRecoveryState", "type": "int8", "versions": "6+", "default": "0",
- "about": "1 if the partition is recovering from an unclean leader election; 0 otherwise." }
- ]}
- ]
+ "validVersions": "none"
Review Comment:
Good catch, I somehow missed this file.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18497:
URL: https://github.com/apache/kafka/pull/18497#issuecomment-2603122568
Assuming the tests look good, I'll merge this. If @jolshan, @cmccabe and/or @jsancio have any comments (for the questions I asked or otherwise), I'll address them in a follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921658142
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -355,28 +362,9 @@ private void generateHeaderVersion(String type) {
buffer.decrementIndent();
continue;
}
- if (type.equals("request") && apiKey == 7) {
- buffer.printf("// Version 0 of ControlledShutdownRequest has a non-standard request header%n");
Review Comment:
We don't need the special case for ControlledShutdown v0 anymore (since it's been removed).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921669453
##########
clients/src/test/java/org/apache/kafka/common/protocol/ProtocolTest.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.
+ */
+package org.apache.kafka.common.protocol;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class ProtocolTest {
+
+ @Test
+ public void testToHtml() {
Review Comment:
Surprisingly, `toHtml` had zero tests even though it breaks the build if it throws an exception (as it's called from one of the docs build targets).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922376442
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -355,28 +362,9 @@ private void generateHeaderVersion(String type) {
buffer.decrementIndent();
continue;
}
- if (type.equals("request") && apiKey == 7) {
- buffer.printf("// Version 0 of ControlledShutdownRequest has a non-standard request header%n");
- buffer.printf("// which does not include clientId. Version 1 of ControlledShutdownRequest%n");
- buffer.printf("// and later use the standard request header.%n");
- buffer.printf("if (_version == 0) {%n");
- buffer.incrementIndent();
- buffer.printf("return (short) 0;%n");
- buffer.decrementIndent();
- buffer.printf("}%n");
- }
- ApiData data = entry.getValue();
- MessageSpec spec;
- if (type.equals("request")) {
- spec = data.requestSpec;
- } else if (type.equals("response")) {
- spec = data.responseSpec;
- } else {
- throw new RuntimeException("Invalid type " + type + " for generateHeaderVersion");
- }
- if (spec == null) {
- throw new RuntimeException("failed to find " + type + " for API key " + apiKey);
- }
+ MessageSpec spec = messageSpec(type, apiKey, entry.getValue());
+ if (!spec.hasValidVersion())
+ continue;
Review Comment:
It seems we need to add `buffer.decrementIndent()` to avoid generating following code.
```java
case 4: // LeaderAndIsr
case 5: // StopReplica
case 6: // UpdateMetadata
case 7: // ControlledShutdown
case 8: // OffsetCommit
```
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -355,28 +362,9 @@ private void generateHeaderVersion(String type) {
buffer.decrementIndent();
continue;
}
- if (type.equals("request") && apiKey == 7) {
- buffer.printf("// Version 0 of ControlledShutdownRequest has a non-standard request header%n");
- buffer.printf("// which does not include clientId. Version 1 of ControlledShutdownRequest%n");
- buffer.printf("// and later use the standard request header.%n");
- buffer.printf("if (_version == 0) {%n");
- buffer.incrementIndent();
- buffer.printf("return (short) 0;%n");
- buffer.decrementIndent();
- buffer.printf("}%n");
- }
- ApiData data = entry.getValue();
- MessageSpec spec;
- if (type.equals("request")) {
- spec = data.requestSpec;
- } else if (type.equals("response")) {
- spec = data.responseSpec;
- } else {
- throw new RuntimeException("Invalid type " + type + " for generateHeaderVersion");
- }
- if (spec == null) {
- throw new RuntimeException("failed to find " + type + " for API key " + apiKey);
- }
+ MessageSpec spec = messageSpec(type, apiKey, entry.getValue());
+ if (!spec.hasValidVersion())
+ continue;
Review Comment:
Or maybe we should skip those api kips to throw `UnsupportedVersionException` directly?
##########
clients/src/main/resources/common/message/LeaderAndIsrRequest.json:
##########
@@ -16,84 +16,6 @@
{
"apiKey": 4,
"type": "request",
- "listeners": ["zkBroker"],
"name": "LeaderAndIsrRequest",
- // Version 1 adds IsNew.
- //
- // Version 2 adds broker epoch and reorganizes the partitions by topic.
- //
- // Version 3 adds AddingReplicas and RemovingReplicas.
- //
- // Version 4 is the first flexible version.
- //
- // Version 5 adds Topic ID and Type to the TopicStates, as described in KIP-516.
- //
- // Version 6 adds LeaderRecoveryState as described in KIP-704.
- //
- // Version 7 adds KRaft Controller ID field as part of KIP-866
- "validVersions": "0-7",
- "flexibleVersions": "4+",
- "fields": [
- { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The current controller ID." },
- { "name": "isKRaftController", "type": "bool", "versions": "7+", "default": "false",
- "about": "If KRaft controller id is used during migration. See KIP-866." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The current controller epoch." },
- { "name": "BrokerEpoch", "type": "int64", "versions": "2+", "ignorable": true, "default": "-1",
- "about": "The current broker epoch." },
- { "name": "Type", "type": "int8", "versions": "5+",
- "about": "The type that indicates whether all topics are included in the request."},
- { "name": "UngroupedPartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "0-1",
- "about": "The state of each partition, in a v0 or v1 message." },
- // In v0 or v1 requests, each partition is listed alongside its topic name.
- // In v2+ requests, partitions are organized by topic, so that each topic name
- // only needs to be listed once.
- { "name": "TopicStates", "type": "[]LeaderAndIsrTopicState", "versions": "2+",
- "about": "Each topic.", "fields": [
- { "name": "TopicName", "type": "string", "versions": "2+", "entityType": "topicName",
- "about": "The topic name." },
- { "name": "TopicId", "type": "uuid", "versions": "5+", "ignorable": true,
- "about": "The unique topic ID." },
- { "name": "PartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "2+",
- "about": "The state of each partition." }
- ]},
- { "name": "LiveLeaders", "type": "[]LeaderAndIsrLiveLeader", "versions": "0+",
- "about": "The current live leaders.", "fields": [
- { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The leader's broker ID." },
- { "name": "HostName", "type": "string", "versions": "0+",
- "about": "The leader's hostname." },
- { "name": "Port", "type": "int32", "versions": "0+",
- "about": "The leader's port." }
- ]}
- ],
- "commonStructs": [
- { "name": "LeaderAndIsrPartitionState", "versions": "0+", "fields": [
- { "name": "TopicName", "type": "string", "versions": "0-1", "entityType": "topicName", "ignorable": true,
- "about": "The topic name. This is only present in v0 or v1." },
- { "name": "PartitionIndex", "type": "int32", "versions": "0+",
- "about": "The partition index." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The controller epoch." },
- { "name": "Leader", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The broker ID of the leader." },
- { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
- "about": "The leader epoch." },
- { "name": "Isr", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The in-sync replica IDs." },
- { "name": "PartitionEpoch", "type": "int32", "versions": "0+",
- "about": "The current epoch for the partition. The epoch is a monotonically increasing value which is incremented after every partition change. (Since the LeaderAndIsr request is only used by the legacy controller, this corresponds to the zkVersion)." },
- { "name": "Replicas", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The replica IDs." },
- { "name": "AddingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are adding this partition to, or null if no replicas are being added." },
- { "name": "RemovingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are removing this partition from, or null if no replicas are being removed." },
- { "name": "IsNew", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
- "about": "Whether the replica should have existed on the broker or not." },
- { "name": "LeaderRecoveryState", "type": "int8", "versions": "6+", "default": "0",
- "about": "1 if the partition is recovering from an unclean leader election; 0 otherwise." }
- ]}
- ]
+ "validVersions": "none"
Review Comment:
Maybe we can add "This request was removed in Apache Kafka 4.0." to this file?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922781434
##########
core/src/main/scala/kafka/log/LogManager.scala:
##########
@@ -35,7 +35,6 @@ import scala.jdk.CollectionConverters._
import scala.collection._
import scala.collection.mutable.ArrayBuffer
import scala.util.{Failure, Success, Try}
-import org.apache.kafka.common.requests.{AbstractControlRequest, LeaderAndIsrRequest}
Review Comment:
Good point - this can go if we move `Type` to `LeaderAndIsrRequest`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922579707
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -355,28 +362,9 @@ private void generateHeaderVersion(String type) {
buffer.decrementIndent();
continue;
}
- if (type.equals("request") && apiKey == 7) {
- buffer.printf("// Version 0 of ControlledShutdownRequest has a non-standard request header%n");
- buffer.printf("// which does not include clientId. Version 1 of ControlledShutdownRequest%n");
- buffer.printf("// and later use the standard request header.%n");
- buffer.printf("if (_version == 0) {%n");
- buffer.incrementIndent();
- buffer.printf("return (short) 0;%n");
- buffer.decrementIndent();
- buffer.printf("}%n");
- }
- ApiData data = entry.getValue();
- MessageSpec spec;
- if (type.equals("request")) {
- spec = data.requestSpec;
- } else if (type.equals("response")) {
- spec = data.responseSpec;
- } else {
- throw new RuntimeException("Invalid type " + type + " for generateHeaderVersion");
- }
- if (spec == null) {
- throw new RuntimeException("failed to find " + type + " for API key " + apiKey);
- }
+ MessageSpec spec = messageSpec(type, apiKey, entry.getValue());
+ if (!spec.hasValidVersion())
+ continue;
Review Comment:
Yeah, the intent was to skip them - this is a bug. I'll fix it and add a test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma merged PR #18497:
URL: https://github.com/apache/kafka/pull/18497
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18497:
URL: https://github.com/apache/kafka/pull/18497#issuecomment-2601516745
@chia7712 @mumrah Can one of you please review this? It would be good to get this into 4.0 so that the protocol apis are correctly specified and documented.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921659893
##########
core/src/test/scala/unit/kafka/server/ShareGroupHeartbeatRequestTest.scala:
##########
@@ -894,7 +894,7 @@ class ShareGroupHeartbeatRequestTest(cluster: ClusterInstance) {
val allPartitionsMetadata = waitForAllPartitionsMetadata(brokersToValidate, topic, totalPartitionCount)
(0 until totalPartitionCount - 1).foreach(i => {
allPartitionsMetadata.get(new TopicPartition(topic, i)).foreach { partitionMetadata =>
- assertEquals(totalPartitionCount, partitionMetadata.replicas.size)
+ assertEquals(totalPartitionCount, partitionMetadata.isr.size)
Review Comment:
This is not exactly the same behavior, but the updated behavior is fine for the test (since isr and replicas is the same). The reason for the change is that the replicas are no longer returned by the method being used here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921670123
##########
clients/src/main/resources/common/message/RequestHeader.json:
##########
@@ -36,7 +38,7 @@
// ApiVersionsRequest, even if it is from a newer version.
// Since the client is sending the ApiVersionsRequest in order to discover what
// versions are supported, the client does not know the best version to use.
- { "name": "ClientId", "type": "string", "versions": "1+", "nullableVersions": "1+", "ignorable": true,
Review Comment:
Removed `ignorable` since the `ClientId` is always set now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921654536
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -2689,157 +2680,6 @@ class ReplicaManagerTest {
}
}
- @Test
- def testFullLeaderAndIsrStrayPartitions(): Unit = {
- val logManager = TestUtils.createLogManager(config.logDirs.map(new File(_)), defaultConfig = new LogConfig(new Properties()), time = time)
- val quotaManager = QuotaFactory.instantiate(config, metrics, time, "")
- val replicaManager = new ReplicaManager(
- metrics = metrics,
- config = config,
- time = time,
- scheduler = time.scheduler,
- logManager = logManager,
- quotaManagers = quotaManager,
- metadataCache = MetadataCache.kRaftMetadataCache(config.brokerId, () => KRaftVersion.KRAFT_VERSION_0),
- logDirFailureChannel = new LogDirFailureChannel(config.logDirs.size),
- alterPartitionManager = alterPartitionManager,
- threadNamePrefix = Option(this.getClass.getName))
-
- try {
-
- logManager.startup(Set.empty[String])
-
- // Create a hosted topic, a hosted topic that will become stray
- createHostedLogs("hosted-topic", numLogs = 2, replicaManager).toSet
- createHostedLogs("hosted-stray", numLogs = 10, replicaManager).toSet
-
- val lisr = new LeaderAndIsrRequest.Builder(ApiKeys.LEADER_AND_ISR.latestVersion,
- 3000, 0, brokerEpoch,
- Seq(
- new LeaderAndIsrPartitionState()
- .setTopicName("hosted-topic")
- .setPartitionIndex(0)
- .setControllerEpoch(controllerEpoch)
- .setLeader(0)
- .setLeaderEpoch(10)
- .setIsr(Seq[Integer](0, 1).asJava)
- .setPartitionEpoch(0)
- .setReplicas(Seq[Integer](0, 1).asJava)
- .setIsNew(false),
- new LeaderAndIsrPartitionState()
- .setTopicName("hosted-topic")
- .setPartitionIndex(1)
- .setControllerEpoch(controllerEpoch)
- .setLeader(1)
- .setLeaderEpoch(10)
- .setIsr(Seq[Integer](1, 0).asJava)
- .setPartitionEpoch(0)
- .setReplicas(Seq[Integer](1, 0).asJava)
- .setIsNew(false)
- ).asJava,
- topicIds.asJava,
- Set(new Node(0, "host0", 0), new Node(1, "host1", 1)).asJava,
- true,
- AbstractControlRequest.Type.FULL
- ).build()
-
- replicaManager.becomeLeaderOrFollower(0, lisr, (_, _) => ())
-
- val ht0 = replicaManager.getPartition(new TopicPartition("hosted-topic", 0))
- assertTrue(ht0.isInstanceOf[HostedPartition.Online])
-
- val stray0 = replicaManager.getPartition(new TopicPartition("hosted-stray", 0))
-
- assertTrue(stray0.isInstanceOf[HostedPartition.Online])
- } finally {
- Utils.tryAll(util.Arrays.asList[Callable[Void]] (
- () => {
- replicaManager.shutdown(checkpointHW = false)
- null
- },
- () => {
- logManager.shutdown()
- null
- },
- () => {
- quotaManager.shutdown()
- null
- }
- ))
- }
- }
-
- @Test
- def testUpdateStrayLogs(): Unit = {
Review Comment:
`ReplicaManager.updateStrayLogs` was deleted and hence this method is no longer needed. @cmccabe @jsancio Can you please confirm that we have tests for the kraft flow of stray log management?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "jolshan (via GitHub)" <gi...@apache.org>.
jolshan commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1924041096
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4732,67 +4549,6 @@ class ReplicaManagerTest {
}
}
- @Test
- def testPartitionMetadataFileCreatedAfterPreviousRequestWithoutIds(): Unit = {
Review Comment:
I think your conclusion makes sense.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921866074
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4843,53 +4599,39 @@ class ReplicaManagerTest {
try {
val brokerList = Seq[Integer](0, 1).asJava
val topicPartition = new TopicPartition(topic, 0)
- val topicPartitionFoo = new TopicPartition("foo", 0)
val topicPartitionFake = new TopicPartition("fakeTopic", 0)
val topicIds = Map(topic -> Uuid.ZERO_UUID, "foo" -> Uuid.randomUuid()).asJava
val topicNames = topicIds.asScala.map(_.swap).asJava
- def leaderAndIsrRequest(epoch: Int, name: String, version: Short): LeaderAndIsrRequest = LeaderAndIsrRequest.parse(
- new LeaderAndIsrRequest.Builder(version, 0, 0, brokerEpoch,
- Seq(new LeaderAndIsrPartitionState()
- .setTopicName(name)
- .setPartitionIndex(0)
- .setControllerEpoch(0)
- .setLeader(0)
- .setLeaderEpoch(epoch)
- .setIsr(brokerList)
- .setPartitionEpoch(0)
- .setReplicas(brokerList)
- .setIsNew(true)).asJava,
- topicIds,
- Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build().serialize(), version)
+ def leaderAndIsrRequest(epoch: Int, name: String): LeaderAndIsrRequest =
+ new LeaderAndIsrRequest.Builder(0, 0, brokerEpoch,
+ Seq(new LeaderAndIsrRequest.PartitionState()
+ .setTopicName(name)
+ .setPartitionIndex(0)
+ .setControllerEpoch(0)
+ .setLeader(0)
+ .setLeaderEpoch(epoch)
+ .setIsr(brokerList)
+ .setPartitionEpoch(0)
+ .setReplicas(brokerList)
+ .setIsNew(true)).asJava,
+ topicIds,
+ Set(new Node(0, "host1", 0), new Node(1, "host2", 1)).asJava).build()
// There is no file if the topic does not have an associated topic ID.
- val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic", ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
+ val response = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, "fakeTopic"), (_, _) => ())
assertTrue(replicaManager.localLog(topicPartitionFake).isDefined)
val log = replicaManager.localLog(topicPartitionFake).get
assertFalse(log.partitionMetadataFile.get.exists())
assertEquals(Errors.NONE, response.partitionErrors(topicNames).get(topicPartition))
// There is no file if the topic has the default UUID.
- val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic, ApiKeys.LEADER_AND_ISR.latestVersion), (_, _) => ())
+ val response2 = replicaManager.becomeLeaderOrFollower(0, leaderAndIsrRequest(0, topic), (_, _) => ())
assertTrue(replicaManager.localLog(topicPartition).isDefined)
val log2 = replicaManager.localLog(topicPartition).get
assertFalse(log2.partitionMetadataFile.get.exists())
assertEquals(Errors.NONE, response2.partitionErrors(topicNames).get(topicPartition))
- // There is no file if the request an older version
Review Comment:
These removed lines are no longer relevant.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921722811
##########
core/src/main/scala/kafka/server/ReplicaManager.scala:
##########
@@ -419,42 +418,6 @@ class ReplicaManager(val config: KafkaConfig,
brokerTopicStats.removeMetrics(topic)
}
- private[server] def updateStrayLogs(strayPartitions: Iterable[TopicPartition]): Unit = {
Review Comment:
Only used by tests.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921669453
##########
clients/src/test/java/org/apache/kafka/common/protocol/ProtocolTest.java:
##########
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF 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.
+ */
+package org.apache.kafka.common.protocol;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+
+public class ProtocolTest {
+
+ @Test
+ public void testToHtml() {
Review Comment:
Surprisingly, `toHtml` had zero tests even though it beaks the build if it throws an exception (as it's called from one of the docs build targets).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions (WIP) [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1921655537
##########
core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala:
##########
@@ -4732,67 +4549,6 @@ class ReplicaManagerTest {
}
}
- @Test
- def testPartitionMetadataFileCreatedAfterPreviousRequestWithoutIds(): Unit = {
Review Comment:
@jolshan I believe this and`testPartitionFetchStateUpdatesWithTopicIdChanges` are no longer necessary since KRaft always requires topic ids to be set and 4.0 only supports upgrades from clusters already running in kraft mode. Do you agree or are there reasons to try to make these tests work?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922647268
##########
generator/src/main/java/org/apache/kafka/message/ApiMessageTypeGenerator.java:
##########
@@ -355,28 +362,9 @@ private void generateHeaderVersion(String type) {
buffer.decrementIndent();
continue;
}
- if (type.equals("request") && apiKey == 7) {
- buffer.printf("// Version 0 of ControlledShutdownRequest has a non-standard request header%n");
- buffer.printf("// which does not include clientId. Version 1 of ControlledShutdownRequest%n");
- buffer.printf("// and later use the standard request header.%n");
- buffer.printf("if (_version == 0) {%n");
- buffer.incrementIndent();
- buffer.printf("return (short) 0;%n");
- buffer.decrementIndent();
- buffer.printf("}%n");
- }
- ApiData data = entry.getValue();
- MessageSpec spec;
- if (type.equals("request")) {
- spec = data.requestSpec;
- } else if (type.equals("response")) {
- spec = data.responseSpec;
- } else {
- throw new RuntimeException("Invalid type " + type + " for generateHeaderVersion");
- }
- if (spec == null) {
- throw new RuntimeException("failed to find " + type + " for API key " + apiKey);
- }
+ MessageSpec spec = messageSpec(type, apiKey, entry.getValue());
+ if (!spec.hasValidVersion())
+ continue;
Review Comment:
Done.
##########
clients/src/main/resources/common/message/LeaderAndIsrRequest.json:
##########
@@ -16,84 +16,6 @@
{
"apiKey": 4,
"type": "request",
- "listeners": ["zkBroker"],
"name": "LeaderAndIsrRequest",
- // Version 1 adds IsNew.
- //
- // Version 2 adds broker epoch and reorganizes the partitions by topic.
- //
- // Version 3 adds AddingReplicas and RemovingReplicas.
- //
- // Version 4 is the first flexible version.
- //
- // Version 5 adds Topic ID and Type to the TopicStates, as described in KIP-516.
- //
- // Version 6 adds LeaderRecoveryState as described in KIP-704.
- //
- // Version 7 adds KRaft Controller ID field as part of KIP-866
- "validVersions": "0-7",
- "flexibleVersions": "4+",
- "fields": [
- { "name": "ControllerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The current controller ID." },
- { "name": "isKRaftController", "type": "bool", "versions": "7+", "default": "false",
- "about": "If KRaft controller id is used during migration. See KIP-866." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The current controller epoch." },
- { "name": "BrokerEpoch", "type": "int64", "versions": "2+", "ignorable": true, "default": "-1",
- "about": "The current broker epoch." },
- { "name": "Type", "type": "int8", "versions": "5+",
- "about": "The type that indicates whether all topics are included in the request."},
- { "name": "UngroupedPartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "0-1",
- "about": "The state of each partition, in a v0 or v1 message." },
- // In v0 or v1 requests, each partition is listed alongside its topic name.
- // In v2+ requests, partitions are organized by topic, so that each topic name
- // only needs to be listed once.
- { "name": "TopicStates", "type": "[]LeaderAndIsrTopicState", "versions": "2+",
- "about": "Each topic.", "fields": [
- { "name": "TopicName", "type": "string", "versions": "2+", "entityType": "topicName",
- "about": "The topic name." },
- { "name": "TopicId", "type": "uuid", "versions": "5+", "ignorable": true,
- "about": "The unique topic ID." },
- { "name": "PartitionStates", "type": "[]LeaderAndIsrPartitionState", "versions": "2+",
- "about": "The state of each partition." }
- ]},
- { "name": "LiveLeaders", "type": "[]LeaderAndIsrLiveLeader", "versions": "0+",
- "about": "The current live leaders.", "fields": [
- { "name": "BrokerId", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The leader's broker ID." },
- { "name": "HostName", "type": "string", "versions": "0+",
- "about": "The leader's hostname." },
- { "name": "Port", "type": "int32", "versions": "0+",
- "about": "The leader's port." }
- ]}
- ],
- "commonStructs": [
- { "name": "LeaderAndIsrPartitionState", "versions": "0+", "fields": [
- { "name": "TopicName", "type": "string", "versions": "0-1", "entityType": "topicName", "ignorable": true,
- "about": "The topic name. This is only present in v0 or v1." },
- { "name": "PartitionIndex", "type": "int32", "versions": "0+",
- "about": "The partition index." },
- { "name": "ControllerEpoch", "type": "int32", "versions": "0+",
- "about": "The controller epoch." },
- { "name": "Leader", "type": "int32", "versions": "0+", "entityType": "brokerId",
- "about": "The broker ID of the leader." },
- { "name": "LeaderEpoch", "type": "int32", "versions": "0+",
- "about": "The leader epoch." },
- { "name": "Isr", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The in-sync replica IDs." },
- { "name": "PartitionEpoch", "type": "int32", "versions": "0+",
- "about": "The current epoch for the partition. The epoch is a monotonically increasing value which is incremented after every partition change. (Since the LeaderAndIsr request is only used by the legacy controller, this corresponds to the zkVersion)." },
- { "name": "Replicas", "type": "[]int32", "versions": "0+", "entityType": "brokerId",
- "about": "The replica IDs." },
- { "name": "AddingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are adding this partition to, or null if no replicas are being added." },
- { "name": "RemovingReplicas", "type": "[]int32", "versions": "3+", "ignorable": true, "entityType": "brokerId",
- "about": "The replica IDs that we are removing this partition from, or null if no replicas are being removed." },
- { "name": "IsNew", "type": "bool", "versions": "1+", "default": "false", "ignorable": true,
- "about": "Whether the replica should have existed on the broker or not." },
- { "name": "LeaderRecoveryState", "type": "int8", "versions": "6+", "default": "0",
- "about": "1 if the partition is recovering from an unclean leader election; 0 otherwise." }
- ]}
- ]
+ "validVersions": "none"
Review Comment:
Done.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on PR #18497:
URL: https://github.com/apache/kafka/pull/18497#issuecomment-2602852971
@chia7712 Thanks for the review. Pushed a commit that addresses the comments. Also merged master to this branch for the flaky test fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-14552: Assume a baseline of 3.0 for server protocol versions [kafka]
Posted by "ijuma (via GitHub)" <gi...@apache.org>.
ijuma commented on code in PR #18497:
URL: https://github.com/apache/kafka/pull/18497#discussion_r1922780149
##########
clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java:
##########
@@ -144,10 +151,12 @@ public static RequestHeader parse(ByteBuffer buffer) {
header.size = Math.max(buffer.position() - bufferStartPositionForHeader, 0);
return header;
} catch (UnsupportedVersionException e) {
- throw new InvalidRequestException("Unknown API key " + apiKey, e);
+ throw new InvalidRequestException("Unknown API key " + apiKeyId, e);
} catch (Throwable ex) {
- throw new InvalidRequestException("Error parsing request header. Our best guess of the apiKey is: " +
- apiKey, ex);
+ if (ex instanceof InvalidRequestException)
Review Comment:
Makes sense - not sure why I didn't write that in the first place!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org