I’m getting “Topic not present in metadata after 60000 ms” message on some computers

Here’s my program

package kafkaConsumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.lang.management.ManagementFactory;
import java.lang.management.RuntimeMXBean;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.Scanner;
import java.util.concurrent.Future;

public class KafkaConsumerExample {
    private final static String INTOPIC = "my-intopic";
    private final static String OUTTOPIC = "my-outtopic";
    private final static String BOOTSTRAP_SERVERS = "192.168.10.10:9092";

    private static Producer<Long, String> createProducer(String bootstrapServers) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaProducerExample");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        return new KafkaProducer<>(props);
    }

    private static Consumer<Long, String> createConsumer(String intopic, String bootstrapServers) {
        final Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "KafkaConsumerExample");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "1");
        // Create the consumer using props.
        final Consumer<Long, String> consumer = new KafkaConsumer<>(props);
        // Subscribe to the topic.
        consumer.subscribe(Collections.singletonList(intopic));
        return consumer;
    }

    static boolean run(
            Consumer<Long, String> consumer, Producer<Long, String> producer,
            String inTopic, String outTopic) throws InterruptedException {

        String valueToSend;
        long keyToUse;

        if (consumer == null) {
            Scanner sc = new Scanner(System.in);
            System.out.print("Enter key> ");
            keyToUse = sc.nextLong();
            valueToSend = sc.nextLine();
            System.out.print("Enter value> ");
            valueToSend = sc.nextLine();
        } else {
            Duration delta = Duration.ofSeconds(1);
            ConsumerRecords<Long, String> consumerRecords = consumer.poll(delta);
            while (consumerRecords.count() == 0) {
                consumerRecords = consumer.poll(delta);
            }
            ConsumerRecord<Long, String> record = consumerRecords.iterator().next();
            keyToUse = record.key();
            valueToSend = record.value();
            if (producer != null)
                System.out.println("Got key = " + keyToUse + " and value = " + valueToSend);
        }

        if (producer == null) {
            System.out.println("key = " + keyToUse + " and value = " + valueToSend);
        } else {
            try {
                System.out.println("Creating ProducerRecord");
                final ProducerRecord<Long, String> record =
                        new ProducerRecord<>(outTopic, keyToUse, valueToSend);
                System.out.println("Calling producer.send");
                Future<RecordMetadata> sent = producer.send(record);
                System.out.println("Calling sent.get");
                RecordMetadata metadata = sent.get();
                System.out.println("Calling flush");
                producer.flush();
                System.out.println("After flush");
            } catch (Exception e) {
                System.out.println("Exception sending message: " + e.getMessage());
            }
        }
        return !valueToSend.equals("STOP");
    }

    public static void usage() {
        System.out.println(System.getProperty("sun.java.command"));
        System.out.println();
        System.out.println("Usage parameters: [--intopic name] [--outtopic name] [--bootstrap-servers servers]");
        System.exit(1);
    }

    public static void main(String... args) throws Exception {
        String inTopic = INTOPIC;
        String outTopic = OUTTOPIC;
        String bootstrapServers = BOOTSTRAP_SERVERS;

        for (int i = 0; i < args.length; ++i) {
            if (args[i].equals("--intopic")) {
                if (i == args.length - 1) {
                    usage();
                }
                inTopic = args[++i];
            } else if (args[i].equals("--outtopic")) {
                if (i == args.length - 1) {
                    usage();
                }
                outTopic = args[++i];
            } else if (args[i].equals("--bootstrap-servers")) {
                if (i == args.length - 1) {
                    usage();
                }
                bootstrapServers = args[++i];
            } else {
                usage();
            }
        }

        final Consumer<Long, String> consumer;
        if (inTopic.equals("stdin")) {
            consumer = null;
        } else {
            consumer = createConsumer(inTopic, bootstrapServers);
        }
        final Producer<Long, String> producer;
        if (outTopic.equals("stdout")) {
            producer = null;
        } else {
            producer = createProducer(bootstrapServers);
        }

        while (true) {
            if (!run(consumer, producer, inTopic, outTopic)) {
                break;
            }
        }
        if (consumer != null)
            consumer.close();
        if (producer != null)
            producer.close();
    }
}

I run it on Windows and Linux. On some computers it runs fine, but on other computers, specifically a Linux machine which is not the kafka machine, it consistently gives me this error:

Exception sending message: org.apache.kafka.common.errors.TimeoutException: Topic outtopic not present in metadata after 60000 ms.

This happens, of course, when trying to send a message in the run() function, specifically in the sentence RecordMetadata metadata = sent.get().

This kafka installation allows creation of new topics automatically. In fact, if I enter a new name in the –outtopic parameter, even when sending the message fails, the topic is created.

Any clues why? What am I missing in the configuration?

Thank you

Shimon

Answer

192.168.10.10:9092

This seems to be an internal IP. Check if the clients where you cannot access are within its network range i.e. whether they can access this IP.

Try doing a telnet from your client machine..

telnet 192.168.10.10 9092

If you are not able to telnet then give the IP which can be accessed by your clients and ensure the same in the advertised.listeners also.

Also check your advertised.listeners config. When we connect to a url given in the bootstrap.servers that typically should par with those in the advertised.listeners configuration.

Topic metadata not present means that your client is unable to fetch any information about the given topic i.e. it cannot get metadata through the given bootstrap.servers property.

Advertisement

Answer