kafka AdminClient API等待节点分配时超时

Question

我是Kafka的新手，正在尝试使用AdminClientAPI来管理运行在本地机器上的Kafka服务器。我的设置与Kafka文档的快速启动部分完全相同。唯一的区别是我没有创建任何主题。

我在这个设置上运行任何shell脚本都没有问题，但当我尝试运行以下java代码时：

public class ProducerMain{
public static void main(String[] args) {
Properties props = new Properties();
props.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, 
"localhost:9092");

try(final AdminClient adminClient = 
KafkaAdminClient.create(props)){
try {
final NewTopic newTopic = new NewTopic("test", 1, 
(short)1);
final CreateTopicsResult createTopicsResult = 
adminClient.createTopics( 
Collections.singleton(newTopic));
createTopicsResult.all().get();
}catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
}
}
}

错误：TimeoutException: Timed out waiting for a node assignment

Exception in thread "main" java.lang.RuntimeException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.
at ProducerMain.main(ProducerMain.java:41)
<br>Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.
at org.apache.kafka.common.internals.KafkaFutureImpl.wrapAndThrow(KafkaFutureImpl.java:45)
at org.apache.kafka.common.internals.KafkaFutureImpl.access$000(KafkaFutureImpl.java:32)
at org.apache.kafka.common.internals.KafkaFutureImpl$SingleWaiter.await(KafkaFutureImpl.java:89)
at org.apache.kafka.common.internals.KafkaFutureImpl.get(KafkaFutureImpl.java:258)
at ProducerMain.main(ProducerMain.java:38)
<br>Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

我在网上搜索了可能是什么问题，但到目前为止没有发现任何迹象。任何建议都是受欢迎的，因为我已经穷途末路了。

Answer 1

听起来你的经纪人不健康。。。

此代码运行良好

public class Main {
static final Logger logger = LoggerFactory.getLogger(Main.class);
public static void main(String[] args) {
Properties properties = new Properties();
properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
properties.setProperty(AdminClientConfig.CLIENT_ID_CONFIG, "local-test");
properties.setProperty(AdminClientConfig.RETRIES_CONFIG, "3");
try (AdminClient client = AdminClient.create(properties)) {
final CreateTopicsResult res = client.createTopics(
Collections.singletonList(
new NewTopic("foo", 1, (short) 1)
)
);
res.all().get(5, TimeUnit.SECONDS);
} catch (InterruptedException | ExecutionException | TimeoutException e) {
logger.error("unable to create topic", e);
}
}
}

我可以在经纪人日志中看到，这个主题是创建的

Answer 2

我用bitnami/kafka启动了kafka服务，但得到了完全相同的错误。尝试通过这个版本启动kafka，它可以：https://hub.docker.com/r/wurstmeister/kafka

$ docker run -d --name zookeeper-server --network app-tier 
-e ALLOW_ANONYMOUS_LOGIN=yes  -p 2181:2181 zookeeper:3.6.2
$ docker run -d --name kafka-server --network app-tier --publish 9092:9092 
--env KAFKA_ZOOKEEPER_CONNECT=zookeeper-server:2181 
--env KAFKA_ADVERTISED_HOST_NAME=30.225.51.235 
--env KAFKA_ADVERTISED_PORT=9092  
wurstmeister/kafka

30.225.51.235是主机的ip地址。