如何将OpenShift上的KSQLDB-Cluster连接到本地kerberize Kafka集群

Question

我想实现的：
我们有一个本地Kafka集群。我想在OpenShift中设置KSQLDB并将其连接到本地Kafka集群的代理。

问题：
当我尝试使用命令"/usr/bin/ksql-server-start /etc/ksqldb/ksql-server.properties"启动KSQLDB服务器时，我得到错误消息：

[2020-05-14 15:47:48,519] ERROR Failed to start KSQL (io.confluent.ksql.rest.server.KsqlServerMain:60)
io.confluent.ksql.util.KsqlServerException: Could not get Kafka cluster configuration!
        at io.confluent.ksql.services.KafkaClusterUtil.getConfig(KafkaClusterUtil.java:90)
        at io.confluent.ksql.security.KsqlAuthorizationValidatorFactory.isKafkaAuthorizerEnabled(KsqlAuthorizationValidatorFactory.java:81)
        at io.confluent.ksql.security.KsqlAuthorizationValidatorFactory.create(KsqlAuthorizationValidatorFactory.java:51)
        at io.confluent.ksql.rest.server.KsqlRestApplication.buildApplication(KsqlRestApplication.java:624)
        at io.confluent.ksql.rest.server.KsqlRestApplication.buildApplication(KsqlRestApplication.java:544)
        at io.confluent.ksql.rest.server.KsqlServerMain.createExecutable(KsqlServerMain.java:98)
        at io.confluent.ksql.rest.server.KsqlServerMain.main(KsqlServerMain.java:56)
Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1589471268517) timed out at 1589471268518 after 1 attempt(s)
        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:260)
        at io.confluent.ksql.services.KafkaClusterUtil.getConfig(KafkaClusterUtil.java:60)
        ... 6 more
Caused by: org.apache.kafka.common.errors.TimeoutException: Call(callName=listNodes, deadlineMs=1589471268517) timed out at 1589471268518 after 1 attempt(s)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.

我的配置：
我根据此图像设置Dockerfile：https://hub.docker.com/r/confluentinc/ksqldb-server，端口9092、9093、8080、8082和443打开。

我的服务yaml如下所示：

kind: Service
apiVersion: v1
metadata:
  name: social-media-dev
  namespace: abc
  selfLink: xyz
  uid: xyz
  resourceVersion: '1'
  creationTimestamp: '2020-05-14T09:47:15Z'
  labels:
    app: social-media-dev
  annotations:
    openshift.io/generated-by: OpenShiftNewApp
spec:
  ports:
    - name: social-media-dev
      protocol: TCP
      port: 9092
      targetPort: 9092
      nodePort: 31364
  selector:
    app: social-media-dev
    deploymentconfig: social-media-dev
  clusterIP: XX.XX.XXX.XXX
  type: LoadBalancer
  externalIPs:
    - XXX.XX.XXX.XXX
  sessionAffinity: None
  externalTrafficPolicy: Cluster
status:
  loadBalancer:
    ingress:
      - ip: XX.XX.XXX.XXX

我的ksql服务器。属性文件包括以下信息：
侦听器：http://0.0.0.0:8082
引导。服务器：X.X.X.X:9092，X.X.X.Y:9092，X.X.X.Z:9092

到目前为止我所尝试的：

我试着从我的pod内连接到一个代理，它成功了：
（超时1 bash-c）

我还与监听器一起玩，但随后错误消息变短了，只显示了“无法获得Kafka群集配置！”并且没有超时错误。

我试图将LoadBalancer交换到Nodeport，但也没有成功。

你知道我接下来可以尝试什么吗？

更新：随着Cloudera CDH6的升级，Cloudera Kafka集群现在也可以与Kafka Streams一起工作。因此，我现在能够从Openshift中的KSQLDB集群连接到本地Kafka集群。




Answer 1

更新：随着Cloudera CDH6的升级，Cloudera Kafka集群现在也可以与Kafka Streams一起工作。因此，我现在能够从Openshift中的KSQLDB集群连接到本地Kafka集群。

我还将在这里描述连接到kerberized Kafka集群的最后一种方式，因为我一直在努力使其运行：

1. 获取Kerberos票据并通过SSL建立连接

ksql服务器。属性（它的sasl\u ssl部分）：

security.protocol=SASL_SSL
sasl.mechanism=GSSAPI

ssl.truststore.location=truststore.jks
ssl.truststore.password=password
ssl.truststore.type=JKS

ssl.ca.location=cert

sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="my.keytab" serviceName="kafka"  principal="myprincipal";
serviceName="kafka"

producer.ssl.endpoint.identification.algorithm=HTTPS
producer.security.protocol=SASL_SSL
producer.ssl.truststore.location=truststore.jks
producer.ssl.truststore.password=password
producer.sasl.mechanism=GSSAPI
producer.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="my.keytab" serviceName="kafka"  principal="myprincipal";

consumer.ssl.endpoint.identification.algorithm=HTTPS
consumer.security.protocol=SASL_SSL
consumer.ssl.truststore.location=truststore.jks
consumer.ssl.truststore.password=password
consumer.sasl.mechanism=GSSAPI
consumer.sasl.jaas.config=com.sun.security.auth.module.Krb5LoginModule required useKeyTab=true storeKey=true keyTab="my.keytab" serviceName="kafka"  principal="myprincipal";`

<代码>主机=[主机]-

<代码>主机=[主机]-

<代码>主机=[主机]-