CalvinConfluent commented on code in PR #14465:
URL: https://github.com/apache/kafka/pull/14465#discussion_r1360170744
##########
metadata/src/main/java/org/apache/kafka/controller/ClusterControlManager.java:
##########
@@ -327,6 +328,10 @@ public ControllerResult<BrokerRegistrationReply>
registerBroker(
}
int brokerId = request.brokerId();
BrokerRegistration existing = brokerRegistrations.get(brokerId);
+ if (version == 0 || request.previousBrokerEpoch() != existing.epoch())
{
Review Comment:
This line is more like a placeholder for the following change. Basically, we
will also check if ELR is enabled to adopt the new behaviors.
##########
core/src/test/scala/unit/kafka/log/LogManagerTest.scala:
##########
@@ -127,10 +127,41 @@ class LogManagerTest {
// This should cause log1.close() to fail during LogManger shutdown
sequence.
FileUtils.deleteDirectory(logFile1)
- logManagerForTest.get.shutdown()
+ logManagerForTest.get.shutdown(3)
- assertFalse(Files.exists(new File(logDir1,
LogLoader.CleanShutdownFile).toPath))
- assertTrue(Files.exists(new File(logDir2,
LogLoader.CleanShutdownFile).toPath))
+ assertFalse(Files.exists(new File(logDir1,
CleanShutdownFileHandler.CleanShutdownFileName).toPath))
+ assertTrue(Files.exists(new File(logDir2,
CleanShutdownFileHandler.CleanShutdownFileName).toPath))
+ assertEquals(-1L,
logManagerForTest.get.readBrokerEpochFromCleanShutdownFiles())
+ } finally {
+ logManagerForTest.foreach(manager =>
manager.liveLogDirs.foreach(Utils.delete))
+ }
+ }
+
+ @Test
+ def testCleanShutdownFileWithBrokerEpoch(): Unit = {
+ // We create two directories logDir1 and logDir2 to help effectively test
error handling
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
