背景
我正在尝试基于Kafka设置分布式日志记录系统(我知道像Logstash这样的东西存在...),但我希望能够在之后加入一些Storm拓扑结构,例如在流程变慢时发送通知。
设置
我有一个运行中的服务器(wilfly swarm,使用Keycloack进行身份验证)在端口8082上,它托管了我的日志功能。我可以通过REST将日志行推送到该服务器。在幕后,一个Kafka生产者正在运行并将消息传播到Kafka。
- 我在2181端口上安装了Zookeeper
- 我在9092端口运行代理程序
- 我在8082端口运行日志服务器
我的server.properties文件(针对代理程序):
listeners=PLAINTEXT://localhost:9092
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:Bob;User:Alice;User:anonymous
我的ACL配置:
call kafka\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic
call kafka\bin\windows\kafka-acls.bat --add --allow-principal User:anonymous --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --allow-host http://localhost:8082 --operation Read --operation Write --topic testtopic
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:anonymous --consumer --topic testtopic --group group --allow-host http://localhost:8082
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:anonymous --producer --topic testtopic --allow-host http://localhost:8082
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:anonymous --producer --topic testtopic --allow-host 192.168.3.63
我的(Java)生产者属性:
@Produces
private Producer<String, String> stringStringProducer(){
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("acks", "all");
props.put("retries", 0);
props.put("batch.size", 16384);
props.put("linger.ms", 1);
props.put("buffer.memory", 33554432);
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
Producer<String, String> producer = null;
try {
producer = new KafkaProducer<>(props);
return producer;
} catch (Exception e) {
throw new RuntimeException(e);
}
}
问题
当我尝试通过Java生产者(以及控制台生产者)生成消息时,我会遇到以下错误:
[org.apache.kafka.clients.NetworkClient] (kafka-producer-network-thread | producer-6) Error while fetching metadata with correlation id 10 : {testtopic=UNKNOWN_TOPIC_OR_PARTITION}
有人知道我做错了什么吗?
第一种解决方案
我通过授予127.0.0.1访问权限来解决了这个错误信息:
call kafka\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic testtopic
call kafka\bin\windows\kafka-acls.bat --add --allow-principal User:anonymous --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --allow-host http://localhost:8082 --operation Read --operation Write --topic testtopic
call kafka\bin\windows\kafka-acls.bat --add --allow-principal User:ANONYMOUS --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --allow-host 127.0.0.1 --operation Read --operation Write --topic testtopic
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:anonymous --consumer --topic testtopic --group group --allow-host http://localhost:8082
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:ANONYMOUS --consumer --topic testtopic --group group --allow-host 127.0.0.1
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:anonymous --producer --topic testtopic --allow-host http://localhost:8082
call kafka\bin\windows\kafka-acls.bat --authorizer kafka.security.auth.SimpleAclAuthorizer --authorizer-properties zookeeper.connect=localhost:2181 --add --allow-principal User:ANONYMOUS --producer --topic testtopic --allow-host 127.0.0.1
我通过查看日志文件找到了问题(即转到kafka文件夹中的log4j.properties,并将log4j.logger.kafka.authorizer.logger属性更改为DEBUG。然后您将获得具体错误(即缺少权限)。
新问题:
当我想要生产一条消息时,现在会出现以下情况:
[2017-03-28 15:39:07,704] WARN Error while fetching metadata with correlation id 0 : {testtopic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-03-28 15:39:07,800] WARN Error while fetching metadata with correlation id 1 : {testtopic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-03-28 15:39:07,912] WARN Error while fetching metadata with correlation id 2 : {testtopic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2017-03-28 15:39:08,024] WARN Error while fetching metadata with correlation id 3 : {testtopic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
请问有人知道如何修复这个问题吗?
已解决
我在代理配置(server.properties)中的超级用户中添加了"ANONYMOUS":
authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
super.users=User:Bob;User:Alice;User:ANONYMOUS
#port = 9092
#advertised.host.name = localhost
#listeners=SASL_SSL://localhost:9092
#security.inter.broker.protocol=SASL_SSL
#sasl.mechanism.inter.broker.protocol=PLAIN
#sasl.enabled.mechanisms=PLAIN
host.name=127.0.0.1
advertised.host.name=localhost
advertised.port=9092