Ghe*_*ham 7 java windows apache-kafka kafka-producer-api
我收到错误:
org.apache.kafka.common.errors.TimeoutException: Topic testtopic2 not present in metadata after 60000 ms.
Run Code Online (Sandbox Code Playgroud)
当尝试使用 Java 在 Windows 上的本地 kafka 实例中生成主题时。请注意,主题 testtopic2 存在,我可以使用 Windows 控制台生产者向它生成消息就好了。
在我正在使用的代码下方:
import java.util.Properties;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class Kafka_Producer {
public static void main(String[] args){
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 0);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = new KafkaProducer<String, String>(props);
TestCallback callback = new TestCallback();
for (long i = 0; i < 100 ; i++) {
ProducerRecord<String, String> data = new ProducerRecord<String, String>(
"testtopic2", "key-" + i, "message-"+i );
producer.send(data, callback);
}
producer.close();
}
private static class TestCallback implements Callback {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
System.out.println("Error while producing message to topic :" + recordMetadata);
e.printStackTrace();
} else {
String message = String.format("sent message to topic:%s partition:%s offset:%s", recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset());
System.out.println(message);
}
}
}
}
Run Code Online (Sandbox Code Playgroud)
Pom依赖:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.6.0</version>
</dependency>
Run Code Online (Sandbox Code Playgroud)
Dmi*_*kiy 16
由于目标 Kafka 实例“死亡”或 URL 错误,也可能会出现此错误。
在这种情况下,向 Kafka 发送消息的线程将按时阻塞max.block.ms,默认为 60000 毫秒。
您可以通过传递更改的值来检查是否是由于上述属性所致:
Properties props = new Properties();
...(among others)
props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30000); # 30 sec or any other value of your choice
Run Code Online (Sandbox Code Playgroud)
如果TimeoutException在指定时间之后抛出,那么您应该检查 Kafka 的 URL 是否正确或者 Kafka 实例是否处于活动状态。
Bob*_*bbs 11
我今天遇到了同样的问题。我是 Kafka 的新手,只是试图让示例 Java 生产者和消费者运行。我能够让消费者工作,但一直收到与您一样的“元数据中不存在的主题”错误,与生产者。
最后,出于绝望,我向我的生产者添加了一些代码来转储主题。当我这样做时,由于缺少 jackson-databind 和 jackson-core 包中的类,我出现了运行时错误。添加它们后,我不再收到“主题不存在”错误。我删除了临时添加的主题转储代码,它仍然有效。
首先,我要感谢 Bobb Dobbs 的回答,今天我也为此苦苦挣扎了一段时间。我只想补充一点,我必须添加的唯一依赖项是jackson-databind. 除了 . 之外,这是我的项目中唯一的依赖项kafka-clients。
更新:我对正在发生的事情有了更多的了解。kafka-clients将其依赖项的范围设置jackson-databind为“提供”,这意味着它希望由 JDK 或容器在运行时提供。有关所提供的 Maven 范围的更多详细信息,请参阅本文。
此作用域用于标记 JDK 或容器应在运行时提供的依赖项,因此得名。此范围的一个很好的用例是部署在某个容器中的 Web 应用程序,其中容器本身已经提供了一些库。
我不确定将其范围设置为提供的确切原因,除了这个库可能是人们通常希望自己提供的库,以使其保持最新版本以进行安全修复等。
当我的团队中的某人更改了 spring.kafka.security.protocol 配置的值时(我们在我的项目中使用 Spring),我看到了这个问题。以前在我们的配置中它是“SSL”,但它已更新为PLAINTEXT. 在我们连接到使用 SSL 的集群的较高环境中,我们看到了 OP 遇到的错误。
为什么我们看到这个错误而不是 SSL 错误或身份验证错误,我无法理解,但如果您遇到此错误,可能值得仔细检查您的 Kafka 集群的客户端身份验证配置。
我也遇到了类似的问题,我在我的 MacBook 上的本地环境中尝试了这个。这很令人沮丧,我尝试了几种方法
我在生产中使用Kafka已经超过3年了,但在集群上没有遇到这个问题,只发生在我的本地环境上。然而,重新启动对我来说解决了这个问题。
| 归档时间: |
|
| 查看次数: |
19462 次 |
| 最近记录: |