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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
