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

Code before:
        long refreshBackoffMs = 500L;
        long refreshBackoffMaxMs = 5000L;
        long metadataExpireMs = 60000L;
        long metadataIdleMs = 60000L;
        final Time time = new MockTime();
        final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                new LogContext(), new ClusterResourceListeners(), time);
        final String topic = "topic";
        try (KafkaProducer<String, String> producer = kafkaProducer(configs,
                new StringSerializer(), new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {

            AtomicBoolean running = new AtomicBoolean(true);
            Thread t = new Thread(() -> {
                long startTimeMs = System.currentTimeMillis();
                while (running.get()) {
                    while (!metadata.updateRequested() && System.currentTimeMillis() - startTimeMs < 100)
                        Thread.yield();
                    MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                            singletonMap(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
                    metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                    time.sleep(60 * 1000L);
                }
            });
            t.start();
            Throwable throwable = assertThrows(TimeoutException.class, () -> producer.partitionsFor(topic));
            assertEquals(UnknownTopicOrPartitionException.class, throwable.getCause().getClass());
            running.set(false);
            t.join();
        }
    }

    @Test
    public void testTopicExpiryInMetadata() throws InterruptedException {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
        long refreshBackoffMs = 500L;
        long refreshBackoffMaxMs = 5000L;
        long metadataExpireMs = 60000L;
        long metadataIdleMs = 60000L;
        final Time time = new MockTime();
        final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                new LogContext(), new ClusterResourceListeners(), time);
        final String topic = "topic";
        try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {

            Exchanger<Void> exchanger = new Exchanger<>();

            Thread t = new Thread(() -> {
                try {

Review comment (bug):
what about `assertInstanceOf(UnknownTopicOrPartitionException.class, throwable.getCause());`. Seems clearer and safer (test would fail instead of throwing NPE if cause ends up being null by mistake) 

Code after:
        long refreshBackoffMs = 500L;
        long refreshBackoffMaxMs = 5000L;
        long metadataExpireMs = 60000L;
        long metadataIdleMs = 60000L;
        final Time time = new MockTime();
        final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                new LogContext(), new ClusterResourceListeners(), time);
        final String topic = "topic";
        try (KafkaProducer<String, String> producer = kafkaProducer(configs,
                new StringSerializer(), new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {

            AtomicBoolean running = new AtomicBoolean(true);
            Thread t = new Thread(() -> {
                long startTimeMs = System.currentTimeMillis();
                while (running.get()) {
                    while (!metadata.updateRequested() && System.currentTimeMillis() - startTimeMs < 100)
                        Thread.yield();
                    MetadataResponse updateResponse = RequestTestUtils.metadataUpdateWith("kafka-cluster", 1,
                            singletonMap(topic, Errors.UNKNOWN_TOPIC_OR_PARTITION), emptyMap());
                    metadata.updateWithCurrentRequestVersion(updateResponse, false, time.milliseconds());
                    time.sleep(60 * 1000L);
                }
            });
            t.start();
            Throwable throwable = assertThrows(TimeoutException.class, () -> producer.partitionsFor(topic));
            assertInstanceOf(UnknownTopicOrPartitionException.class, throwable.getCause());
            running.set(false);
            t.join();
        }
    }

    @Test
    public void testTopicNotExistingInMetadata() throws InterruptedException {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
        configs.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, "30000");
        long refreshBackoffMs = 500L;
        long refreshBackoffMaxMs = 5000L;
        long metadataExpireMs = 60000L;
        long metadataIdleMs = 60000L;
        final Time time = new MockTime();
        final ProducerMetadata metadata = new ProducerMetadata(refreshBackoffMs, refreshBackoffMaxMs, metadataExpireMs, metadataIdleMs,
                new LogContext(), new ClusterResourceListeners(), time);
        final String topic = "topic";
        try (KafkaProducer<String, String> producer = kafkaProducer(configs, new StringSerializer(),
                new StringSerializer(), metadata, new MockClient(time, metadata), null, time)) {

            Exchanger<Void> exchanger = new Exchanger<>();

            Thread t = new Thread(() -> {
                try {