@Override
    public void close() {
        close(Duration.ofMillis(Long.MAX_VALUE));
    }
@Test
    void testDeliveryTimeoutAndLingerMsConfig() {
        final Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.CLIENT_ID_CONFIG, "testDeliveryTimeoutAndLingerMsConfig");
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        configs.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 1000);
        configs.put(ProducerConfig.LINGER_MS_CONFIG, 1000);
        configs.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1);

        // delivery.timeout.ms should be equal to or larger than linger.ms + request.timeout.ms
        assertThrows(KafkaException.class, () -> new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer()));

        configs.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, 1000);
        configs.put(ProducerConfig.LINGER_MS_CONFIG, 999);
        configs.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 1);

        assertDoesNotThrow(() -> new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer()).close());
    }