Repository: apache/kafka
File: streams/integration-tests/src/test/java/org/apache/kafka/streams/integration/RestoreIntegrationTest.java (Java)

Code before:
@Tag("integration")
public class RestoreIntegrationTest {
    private static final Logger log = LoggerFactory.getLogger(RestoreIntegrationTest.class);

    private static final Duration RESTORATION_DELAY = Duration.ofMillis(2000);

    private static final int NUM_BROKERS = 2;

    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);
    
    private static Admin admin;

    @BeforeAll
    public static void startCluster() throws IOException {
        CLUSTER.start();
        
        final Properties adminConfig = new Properties();
        adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        admin = Admin.create(adminConfig);
    }

    @AfterAll
    public static void closeCluster() {
        if (admin != null) {
            admin.close();
        }
        CLUSTER.stop();
    }

    private String appId;
    private String inputStream;

    private final int numberOfKeys = 10000;
    private KafkaStreams kafkaStreams;

    private final List<Properties> streamsConfigurations = new ArrayList<>();

    @BeforeEach
    public void createTopics(final TestInfo testInfo) throws InterruptedException {
        appId = safeUniqueTestName(testInfo);
        inputStream = appId + "-input-stream";
        CLUSTER.createTopic(inputStream, 2, 1);
    }

    private Properties props(final boolean stateUpdaterEnabled) {
        return props(mkObjectProperties(mkMap(mkEntry(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled))));
    }

    private Properties props(final Properties extraProperties) {
        final Properties streamsConfiguration = new Properties();


Review comment (refactor):
FYI: there's a utility method in Kafka called `Utils.closeQuietly()` which can make closing resources a one-liner.

Code after:
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.fail;

@Timeout(600)
@Tag("integration")
public class RestoreIntegrationTest {
    private static final Logger log = LoggerFactory.getLogger(RestoreIntegrationTest.class);

    private static final Duration RESTORATION_DELAY = Duration.ofMillis(2000);

    private static final int NUM_BROKERS = 2;

    public static final EmbeddedKafkaCluster CLUSTER = new EmbeddedKafkaCluster(NUM_BROKERS);

    private static Admin admin;

    @BeforeAll
    public static void startCluster() throws IOException {
        CLUSTER.start();
        
        final Properties adminConfig = new Properties();
        adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, CLUSTER.bootstrapServers());
        admin = Admin.create(adminConfig);
    }

    @AfterAll
    public static void closeCluster() {
        Utils.closeQuietly(admin, "admin");
        CLUSTER.stop();
    }

    private String appId;
    private String inputStream;

    private final int numberOfKeys = 10000;
    private KafkaStreams kafkaStreams;

    private final List<Properties> streamsConfigurations = new ArrayList<>();

    @BeforeEach
    public void createTopics(final TestInfo testInfo) throws InterruptedException {
        appId = safeUniqueTestName(testInfo);
        inputStream = appId + "-input-stream";
        CLUSTER.createTopic(inputStream, 2, 1);
    }

    private Properties props(final boolean stateUpdaterEnabled) {
        return props(mkObjectProperties(mkMap(mkEntry(InternalConfig.STATE_UPDATER_ENABLED, stateUpdaterEnabled))));
    }

    private Properties props(final Properties extraProperties) {