wso2 Конечная точка входа - потребитель Kafka

Question

Я создаю потребителя Кафки.

Используя эту статью, я установил Zooker и Kafka. https://dzone.com/articles/running-apache-kafka-on-windows-os

Используя эту документацию, я настраиваю входящую конечную точку Kafka https://docs.wso2.com/display/EI611/Kafka+Inbound+Protocol

Когда я пытаюсь развернуть Inbound Endpoint, я получаю эту ошибку

[2017-09-22 12:19:06,161] [] ERROR - KAFKAPollingConsumer  Error in Creating Kafka Consumer Connector
[2017-09-22 12:19:08,150] []  INFO - KAFKAMessageListener Creating Kafka Consumer Connector...

[2017-09-22 12:19:08,152] [] ОШИБКА - Ошибка KAFKAMessageListener при создании Consumer Connector Kafka.Exception при загрузке контекста входа в JAAS Zookeeper "Клиент" org.apache.kafka.common.KafkaException: Исключение при загрузке входа в JAAS Zookeeper контекст "Клиент"

        at org.apache.kafka.common.security.JaasUtils.isZkSecurityEnabled(JaasUtils.java:43)
        at kafka.consumer.ZookeeperConsumerConnector.connectZk(ZookeeperConsumerConnector.scala:197)
        at kafka.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:142)
        at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:67)
        at kafka.javaapi.consumer.ZookeeperConsumerConnector.<init>(ZookeeperConsumerConnector.scala:70)
        at kafka.consumer.Consumer$.createJavaConsumerConnector(ConsumerConnector.scala:123)
        at kafka.consumer.Consumer.createJavaConsumerConnector(ConsumerConnector.scala)
        at org.wso2.carbon.inbound.endpoint.protocol.kafka.KAFKAMessageListener.createKafkaConsumerConnector(KAFKAMessageListener.java:56)
        at org.wso2.carbon.inbound.endpoint.protocol.kafka.KAFKAPollingConsumer.poll(KAFKAPollingConsumer.java:145)
        at org.wso2.carbon.inbound.endpoint.protocol.kafka.KAFKAPollingConsumer.execute(KAFKAPollingConsumer.java:116)
        at org.wso2.carbon.inbound.endpoint.protocol.kafka.KAFKATask.taskExecute(KAFKATask.java:48)
        at org.wso2.carbon.inbound.endpoint.common.InboundTask.execute(InboundTask.java:45)
        at org.wso2.carbon.mediation.ntask.NTaskAdapter.execute(NTaskAdapter.java:98)
        at org.wso2.carbon.ntask.core.impl.TaskQuartzJobAdapter.execute(TaskQuartzJobAdapter.java:67)
        at org.quartz.core.JobRunShell.run(JobRunShell.java:213)
        at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
        at java.util.concurrent.FutureTask.run(FutureTask.java:266)
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
        at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.SecurityException: java.io.IOException: C:\WS02\WSO2EI~1.1\bin\..\repository\conf\identity\jaas.conf (No such file or directory)
        at sun.security.provider.ConfigFile$Spi.<init>(ConfigFile.java:137)
        at sun.security.provider.ConfigFile.<init>(ConfigFile.java:102)
        at sun.reflect.GeneratedConstructorAccessor74.newInstance(Unknown Source)
        at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
        at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
        at java.lang.Class.newInstance(Class.java:442)
        at javax.security.auth.login.Configuration$2.run(Configuration.java:255)
        at javax.security.auth.login.Configuration$2.run(Configuration.java:247)
        at java.security.AccessController.doPrivileged(Native Method)
        at javax.security.auth.login.Configuration.getConfiguration(Configuration.java:246)
        at org.apache.kafka.common.security.JaasUtils.isZkSecurityEnabled(JaasUtils.java:40)
        ... 19 more

Вызывается: java.io.IOException: C:\WS02\WSO2EI~1.1\bin..\repository\conf\identity\jaas.conf (нет такого файла или каталога)

        at sun.security.provider.ConfigFile$Spi.ioException(ConfigFile.java:666)
        at sun.security.provider.ConfigFile$Spi.init(ConfigFile.java:262)
        at sun.security.provider.ConfigFile$Spi.<init>(ConfigFile.java:135)

Answer 1

Если вам нужно настроить Kafka без безопасности, рекомендуемая версия - 0.8.1.1.

Начиная с версии 0.9 Kafka, они представили механизм безопасности Kafka с SSL и SASL, который будет настраиваться через JAAS. Итак, при настройке Kafka с продуктами WSO2 (EI, DAS, CEP) вам необходимо добавить этот файл конфигурации (jaas.conf) в / repository / conf / identity.

В этом случае продукт WSO2 действует как клиент Kafka, и конфигурация в jaas.conf будет использоваться для входа на сервер kafka. Пожалуйста, следуйте документации по безопасности Kafka [1] для настройки безопасности для сервера и клиента.

Пример содержимого файла jaas.conf, настроенного для SASL/ Plaintext.

KafkaServer {
org.apache.kafka.common.security.plain.PlainLoginModule required
   username="kafka"
   password="kafka-secret"
   user_kafka="kafka-secret"
   user_ibm="ibm-secret";
};

KafkaClient {
org.apache.kafka.common.security.plain.PlainLoginModule required
  username="kafka"
  password="kafka-secret";
};

Обратите внимание, что kafka 0.9 поддерживает только аутентификацию SASL/Kerberos. Пожалуйста, внимательно следуйте документации по безопасности Kafka, связанной с вашей версией Kafka, и настройте ее должным образом.

[1] http://kafka.apache.org/090/documentation.html