root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 2 Packages Projects Releases Wiki Activity

MINOR: ensure LocalLogManagerTestEnv is closed in QuorumControllerTest

This commit is contained in:
Colin P. McCabe 2024-03-22 13:18:59 -07:00
parent 2e8d69b78c
commit ad3876dfdf
1 changed files with 17 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
metadata/src/test/java/org/apache/kafka/controller/QuorumControllerTest.java
View File

@ -1540,15 +1540,14 @@ public class QuorumControllerTest {
@Test
public void testMigrationsEnabledForOldBootstrapMetadataVersion() throws Exception {
try (
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).build()
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).build();
) {
QuorumControllerTestEnv.Builder controlEnvBuilder = new QuorumControllerTestEnv.Builder(logEnv).
setControllerBuilderInitializer(controllerBuilder -> {
controllerBuilder.setZkMigrationEnabled(true);
}).
setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV0, "test"));
QuorumControllerTestEnv controlEnv = controlEnvBuilder.build();
QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
setControllerBuilderInitializer(controllerBuilder -> {
controllerBuilder.setZkMigrationEnabled(true);
}).
setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_3_IV0, "test")).
build();
QuorumController active = controlEnv.activeController();
assertEquals(ZkMigrationState.NONE, active.appendReadEvent("read migration state", OptionalLong.empty(),
() -> active.featureControl().zkMigrationState()).get(30, TimeUnit.SECONDS));
@ -1658,12 +1657,12 @@ public class QuorumControllerTest {
@Test
public void testFailoverDuringMigrationTransaction() throws Exception {
try (
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3).build()
) {
QuorumControllerTestEnv.Builder controlEnvBuilder = new QuorumControllerTestEnv.Builder(logEnv).
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(3).build();
QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
setControllerBuilderInitializer(controllerBuilder -> controllerBuilder.setZkMigrationEnabled(true)).
setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test"));
QuorumControllerTestEnv controlEnv = controlEnvBuilder.build();
setBootstrapMetadata(BootstrapMetadata.fromVersion(MetadataVersion.IBP_3_6_IV1, "test")).
build();
) {
QuorumController active = controlEnv.activeController(true);
ZkRecordConsumer migrationConsumer = active.zkRecordConsumer();
migrationConsumer.beginMigration().get(30, TimeUnit.SECONDS);
@ -1703,18 +1702,17 @@ public class QuorumControllerTest {
@EnumSource(value = MetadataVersion.class, names = {"IBP_3_4_IV0", "IBP_3_5_IV0", "IBP_3_6_IV0", "IBP_3_6_IV1"})
public void testBrokerHeartbeatDuringMigration(MetadataVersion metadataVersion) throws Exception {
try (
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).build()
) {
QuorumControllerTestEnv.Builder controlEnvBuilder = new QuorumControllerTestEnv.Builder(logEnv).
LocalLogManagerTestEnv logEnv = new LocalLogManagerTestEnv.Builder(1).build();
QuorumControllerTestEnv controlEnv = new QuorumControllerTestEnv.Builder(logEnv).
setControllerBuilderInitializer(controllerBuilder ->
controllerBuilder
.setZkMigrationEnabled(true)
.setMaxIdleIntervalNs(OptionalLong.of(TimeUnit.MILLISECONDS.toNanos(100)))
).
setBootstrapMetadata(BootstrapMetadata.fromVersion(metadataVersion, "test"));
QuorumControllerTestEnv controlEnv = controlEnvBuilder.build();
setBootstrapMetadata(BootstrapMetadata.fromVersion(metadataVersion, "test")).
build();
) {
QuorumController active = controlEnv.activeController(true);
// Register a ZK broker
BrokerRegistrationReply reply = active.registerBroker(ANONYMOUS_CONTEXT,
new BrokerRegistrationRequestData().
@ -1753,7 +1751,6 @@ public class QuorumControllerTest {
active.processBrokerHeartbeat(ANONYMOUS_CONTEXT, new BrokerHeartbeatRequestData().
setWantFence(false).setBrokerEpoch(reply.epoch()).setBrokerId(0).
setCurrentMetadataOffset(100100L)).get());
}
}
}