Repository: apache/kafka
File: clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java (Java)

Code before:
        assertTrue(ce.getMessage().contains("not string key"), "Unexpected exception message: " + ce.getMessage());
    }

    @Test
    public void testConstructorWithInvalidMetricReporterClass() {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "an.invalid.class");
        KafkaException ce = assertThrows(
                KafkaException.class,
                () -> new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()));
        assertTrue(ce.getMessage().contains("Failed to construct kafka producer"), "Unexpected exception message: " + ce.getMessage());
        assertTrue(ce.getCause().getMessage().contains("Class an.invalid.class cannot be found"), "Unexpected cause: " + ce.getCause());
    }

    @Test
    public void testSerializerClose() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        configs.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
        configs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL);
        final int oldInitCount = MockSerializer.INIT_COUNT.get();
        final int oldCloseCount = MockSerializer.CLOSE_COUNT.get();

        KafkaProducer<String, String> producer = new KafkaProducer<>(
                configs, new MockSerializer(), new MockSerializer());
        assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
        assertEquals(oldCloseCount, MockSerializer.CLOSE_COUNT.get());

        producer.close();
        assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
        assertEquals(oldCloseCount + 2, MockSerializer.CLOSE_COUNT.get());
    }

    @Test
    public void testInterceptorConstructClose() {
        try {
            Properties props = new Properties();
            // test with client ID assigned by KafkaProducer
            props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
            props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName());
            props.setProperty(MockProducerInterceptor.APPEND_STRING_PROP, "something");

            try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                    props, new StringSerializer(), new StringSerializer())) {
                assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
                assertEquals(0, MockProducerInterceptor.CLOSE_COUNT.get());

                // Cluster metadata will only be updated on calling onSend.
                assertNull(MockProducerInterceptor.CLUSTER_META.get());

Review comment (suggestion):
```java
        try (var ignored = new KafkaProducer<>(configs, new MockSerializer(), new MockSerializer())) {
            assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
            assertEquals(oldCloseCount, MockSerializer.CLOSE_COUNT.get());
        }
```

Code after:
        assertTrue(ce.getMessage().contains("not string key"), "Unexpected exception message: " + ce.getMessage());
    }

    @Test
    public void testConstructorWithInvalidMetricReporterClass() {
        Properties props = new Properties();
        props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        props.setProperty(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, "an.invalid.class");
        KafkaException ce = assertThrows(
                KafkaException.class,
                () -> new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()));
        assertTrue(ce.getMessage().contains("Failed to construct kafka producer"), "Unexpected exception message: " + ce.getMessage());
        assertTrue(ce.getCause().getMessage().contains("Class an.invalid.class cannot be found"), "Unexpected cause: " + ce.getCause());
    }

    @Test
    public void testSerializerClose() {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testConstructorClose");
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        configs.put(ProducerConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
        configs.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, CommonClientConfigs.DEFAULT_SECURITY_PROTOCOL);
        final int oldInitCount = MockSerializer.INIT_COUNT.get();
        final int oldCloseCount = MockSerializer.CLOSE_COUNT.get();

        try (var ignored = new KafkaProducer<>(configs, new MockSerializer(), new MockSerializer())) {
            assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
            assertEquals(oldCloseCount, MockSerializer.CLOSE_COUNT.get());
        }

        assertEquals(oldInitCount + 2, MockSerializer.INIT_COUNT.get());
        assertEquals(oldCloseCount + 2, MockSerializer.CLOSE_COUNT.get());
    }

    @Test
    public void testInterceptorConstructClose() {
        try {
            Properties props = new Properties();
            // test with client ID assigned by KafkaProducer
            props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
            props.setProperty(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, MockProducerInterceptor.class.getName());
            props.setProperty(MockProducerInterceptor.APPEND_STRING_PROP, "something");

            try (KafkaProducer<String, String> producer = new KafkaProducer<>(
                    props, new StringSerializer(), new StringSerializer())) {
                assertEquals(1, MockProducerInterceptor.INIT_COUNT.get());
                assertEquals(0, MockProducerInterceptor.CLOSE_COUNT.get());

                // Cluster metadata will only be updated on calling onSend.
                assertNull(MockProducerInterceptor.CLUSTER_META.get());
            }