上下文
我正在尝试建立一个基于kafka的分布式日志系统(我知道有一些东西像logstash…)但我希望能够在之后放置一些storm拓扑,例如,在流变慢时发送通知。
设置
我在端口8082上有一个正在运行的服务器(wilfly swarm,keycloack authenticated),它承载着我的日志功能。我可以通过rest把日志线推到这个服务器上。在幕后,Kafka制作人正在运行并向Kafka传播这些信息。
2181港有Zookeeper
我有一个经纪人在9092港
我让logserver在8082端口运行
my 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)中向超级用户添加了“匿名”:
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
1条答案
按热度按时间s2j5cfk01#
出现此问题的原因是您在以下行中启用了授权:
但经纪人是和
User:ANONYMOUS
,因为以下行:也就是说,代理无法对自己进行身份验证。在我的情况下(ssl身份验证),我必须执行以下操作:
使用启用代理间安全性
security.inter.broker.protocol=SSL
.通过设置禁用代理的纯文本端口
listeners=SSL://broker1:9092
(请注意,没有PLAINTEXT://broker1:9091
)使用为在我的ssl证书中定义的用户定义acl
kafka-acls.sh
.重新启动代理。
第。s。你的答案中的变通方法是不鼓励的。你可以在这里了解它的含义。