junrao commented on code in PR #14836:
URL: https://github.com/apache/kafka/pull/14836#discussion_r1406820475
##########
core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala:
##########
@@ -201,7 +201,7 @@ class BrokerLifecycleManagerTest {
while (!future.isDone || context.mockClient.hasInFlightRequests) {
context.poll()
manager.eventQueue.wakeup()
- context.time.sleep(100)
+ context.time.sleep(5)
Review Comment:
What's causing the following failure before? Does this change fix the issue?
```
rg.opentest4j.AssertionFailedError:
Expected :Set(h3sC4Yk-Q9-fd0ntJTocCA, ej8Q9_d2Ri6FXNiTxKFiow,
1iF76HVNRPqC7Y4r6647eg)
Actual :Set(h3sC4Yk-Q9-fd0ntJTocCA, ej8Q9_d2Ri6FXNiTxKFiow)
```
##########
core/src/test/scala/unit/kafka/server/BrokerLifecycleManagerTest.scala:
##########
@@ -214,28 +214,20 @@ class BrokerLifecycleManagerTest {
ctx.controllerNodeProvider.node.set(controllerNode)
val registration = prepareResponse(ctx, new BrokerRegistrationResponse(new
BrokerRegistrationResponseData().setBrokerEpoch(1000)))
- val heartbeats = Seq.fill(6)(prepareResponse[BrokerHeartbeatRequest](ctx,
new BrokerHeartbeatResponse(new BrokerHeartbeatResponseData())))
-
manager.start(() => ctx.highestMetadataOffset.get(),
ctx.mockChannelManager, ctx.clusterId, ctx.advertisedListeners,
Collections.emptyMap(), OptionalLong.empty())
poll(ctx, manager, registration)
manager.propagateDirectoryFailure(Uuid.fromString("h3sC4Yk-Q9-fd0ntJTocCA"))
- poll(ctx, manager, heartbeats(0)).data()
- val dirs1 = poll(ctx, manager, heartbeats(1)).data().offlineLogDirs()
-
manager.propagateDirectoryFailure(Uuid.fromString("ej8Q9_d2Ri6FXNiTxKFiow"))
- poll(ctx, manager, heartbeats(2)).data()
- val dirs2 = poll(ctx, manager, heartbeats(3)).data().offlineLogDirs()
-
manager.propagateDirectoryFailure(Uuid.fromString("1iF76HVNRPqC7Y4r6647eg"))
- poll(ctx, manager, heartbeats(4)).data()
- val dirs3 = poll(ctx, manager, heartbeats(5)).data().offlineLogDirs()
-
- assertEquals(Set("h3sC4Yk-Q9-fd0ntJTocCA").map(Uuid.fromString),
dirs1.asScala.toSet)
- assertEquals(Set("h3sC4Yk-Q9-fd0ntJTocCA",
"ej8Q9_d2Ri6FXNiTxKFiow").map(Uuid.fromString), dirs2.asScala.toSet)
- assertEquals(Set("h3sC4Yk-Q9-fd0ntJTocCA", "ej8Q9_d2Ri6FXNiTxKFiow",
"1iF76HVNRPqC7Y4r6647eg").map(Uuid.fromString), dirs3.asScala.toSet)
+ val latestHeartbeat = Seq.fill(10)(
+ prepareResponse[BrokerHeartbeatRequest](ctx, new
BrokerHeartbeatResponse(new BrokerHeartbeatResponseData()))
+ ).map(poll(ctx, manager, _)).last
Review Comment:
Hmm, KIP-858 says "The UUIDs for the newly failed log directories are
included in the BrokerHeartbeat request until the broker receives a successful
response.". How do we guarantee that only the 10th HeartbeatRequest picks up
the failed log dirs?
--
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]
