ssl - 启用 SASL、SSL 或 SASL_SSL 时 GCS 连接器失败

Question

我已经能够在未启用 SASL 或 SSL 的情况下成功连接 GCS 连接器。当我启用 SASL 和 SSL 时； connect-standalone 似乎无法与代理通信。

问题似乎出在 gcs-sink-license-manager 上。这是我从日志中发现的，但它们对我真正找出问题所在并不是很有帮助....

日志

[2018-12-19 16:29:05,645] INFO [AdminClient clientId=gcs-sink-license-manager] Metadata update failed (org.apache.kafka.clients.admin.internals.AdminMetadataManager:238)
org.apache.kafka.common.errors.TimeoutException: Timed out waiting to send the call.
[2018-12-19 16:29:05,647] ERROR WorkerConnector{id=gcs-sink} Error while starting connector (org.apache.kafka.connect.runtime.WorkerConnector:119)
org.apache.kafka.connect.errors.ConnectException: Timed out while checking for or creating topic(s) '_confluent-command'. This could indicate a connectivity issue, unavailable topic partitions, or if this is your first use of the topic it may have taken too long to create.
    at org.apache.kafka.connect.util.TopicAdmin.createTopics(TopicAdmin.java:251)
    at io.confluent.license.LicenseStore$1.run(LicenseStore.java:159)
    at org.apache.kafka.connect.util.KafkaBasedLog.start(KafkaBasedLog.java:126)
    at io.confluent.license.LicenseStore.start(LicenseStore.java:187)
    at io.confluent.license.LicenseManager.<init>(LicenseManager.java:42)
    at io.confluent.connect.gcs.GcsSinkConnector.checkLicense(GcsSinkConnector.java:80)
    at io.confluent.connect.gcs.GcsSinkConnector.start(GcsSinkConnector.java:67)
    at org.apache.kafka.connect.runtime.WorkerConnector.doStart(WorkerConnector.java:111)
    at org.apache.kafka.connect.runtime.WorkerConnector.start(WorkerConnector.java:136)
    at org.apache.kafka.connect.runtime.WorkerConnector.transitionTo(WorkerConnector.java:195)
    at org.apache.kafka.connect.runtime.Worker.startConnector(Worker.java:241)
    at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.startConnector(StandaloneHerder.java:297)
    at org.apache.kafka.connect.runtime.standalone.StandaloneHerder.putConnectorConfig(StandaloneHerder.java:206)
    at org.apache.kafka.connect.cli.ConnectStandalone.main(ConnectStandalone.java:107)
Caused by: org.apache.kafka.common.errors.TimeoutException: Timed out waiting for a node assignment.
[2018-12-19 16:29:05,649] INFO Finished creating connector gcs-sink (org.apache.kafka.connect.runtime.Worker:257)
[2018-12-19 16:29:05,650] INFO Skipping reconfiguration of connector gcs-sink since it is not running (org.apache.kafka.connect.runtime.standalone.StandaloneHerder:329)
[2018-12-19 16:29:05,652] INFO Created connector gcs-sink (org.apache.kafka.connect.cli.ConnectStandalone:104)

连接器属性

connector.class="io.confluent.connect.gcs.GcsSinkConnector"
storage.class="io.confluent.connect.gcs.storage.GcsStorage"
bootstrap.servers=kafka1:19092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
security.protocol=SASL_PLAINTEXT
sasl.mechanism=PLAIN
offset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000
plugin.path=/usr/share/java,/usr/share/confluent-hub-components

gcs.sasl.properties

#Connector
format.class=io.confluent.connect.gcs.format.json.JsonFormat
partitioner.class=io.confluent.connect.storage.partitioner.DefaultPartitioner
flush.size=3
# confluent.license=

#GCS
name=gcs-sink
connector.class=io.confluent.connect.gcs.GcsSinkConnector
gcs.bucket.name=kafka-bucket-4c
gcs.part.size=5242880
gcs.credentials.path=/usr/share/assets/gcs-key.json
confluent.topic.bootstrap.servers=kafka1:19092
topics=sandbox
confluent.topic.replication.factor=1
#Storage
storage.class=io.confluent.connect.gcs.storage.GcsStorage
client.id=gcs-standalone-sink


# Sink authentication settings
consumer.log4j.root.loglevel=DEBUG
consumer.bootstrap.servers=kafka1:19092
consumer.sasl.mechanism=PLAIN
consumer.security.protocol=SASL_PLAINTEXT
consumer.ssl.endpoint.identification.algorithm=

码头文件

FROM confluentinc/cp-kafka-connect

ADD assets /usr/share/assets

# ENV CONNECT_OPTS "-Djava.security.auth.login.config=/usr/share/assets/kafka_admin_account.conf -Djavax.net.ssl.trustStore=/usr/share/assets/secrets/kafka.client.truststore.jks -Djavax.net.ssl.trustStorePassword=changeit"
ENV KAFKA_OPTS "-Djava.security.auth.login.config=/usr/share/assets/secrets/kafka_admin_account.conf -Djavax.net.debug=all"
ENV CONNECT_OPTS "-Djava.security.auth.login.config=/usr/share/assets/secrets/kafka_admin_account.conf -Djavax.net.debug=all"

COPY assets/secrets/cacerts /usr/lib/jvm/zulu-8-amd64/jre/lib/security/cacerts

CMD ["/bin/bash", "-c", "connect-standalone ${CONNECT_PROPS} ${GCS_PROPS}"]

docker-compose 文件

kafka1:
image: company-kafka-secure
# build: ./
depends_on:
- zookeeper
ports:
- 19091:19091
environment:
  KAFKA_BROKER_ID: 1
  KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
  KAFKA_ADVERTISED_LISTENERS: SASL_PLAINTEXT://kafka1:19092,EXT://localhost:19091
  KAFKA_LISTENERS: SASL_PLAINTEXT://:19092,EXT://:19091
  KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: SASL_PLAINTEXT:SASL_PLAINTEXT,EXT:SASL_PLAINTEXT
  KAFKA_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
  KAFKA_SECURITY_INTER_BROKER_PROTOCOL: SASL_PLAINTEXT
  KAFKA_SASL_ENABLED_MECHANISMS: PLAIN
  KAFKA_ZOOKEEPER_CONNECTION_TIMEOUT_MS: 6000
  ZOOKEEPER_SASL_ENABLED: "false"
  KAFKA_AUTHORIZER_CLASS_NAME: com.us.digital.kafka.security.authorization.KafkaAuthorizer
  CONFLUENT_METRICS_ENABLE: "false"
volumes:
- ./secrets:/etc/kafka/secrets
networks:
- message_hub

kafka_gcs_connect:
build: ./kafka-connect
ports:
  - 28082:28082
depends_on:
- kafka1
- kafka3
- kafka2
- zookeeper
environment: 
  CONNECT_PROPS: /usr/share/assets/connect-standalone.sasl.properties
  CONNECT_REST_PORT: 28082
  GCS_PROPS: /usr/share/assets/gcs.sasl.properties
networks:
- message_hub

Answer 1

CONNECT_BOOTSTRAP_SERVERS=kafka1:19092,kafka2:29092,kafka3:39092
CONNECT_CONFLUENT_TOPIC_BOOTSTRAP_SERVERS=kafka1:19092,kafka2:29092,kafka3:39092
CONNECT_CONFLUENT_LICENSE=
CONNECT_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE=false
CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE=false
CONNECT_CONFIG_STORAGE_TOPIC=connect-config
CONNECT_OFFSET_STORAGE_TOPIC=connect-offsets
CONNECT_STATUS_STORAGE_TOPIC=connect-status
CONNECT_REPLICATION_FACTOR=1
CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1
CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1
CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1
CONNECT_SECURITY_PROTOCOL=SASL_PLAINTEXT
CONNECT_SASL_MECHANISM=PLAIN
CONNECT_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=
CONNECT_CONSUMER_BOOTSTRAP_SERVERS=kafka1:19092,kafka2:29092,kafka3:39092
CONNECT_CONSUMER_SECURITY_PROTOCOL=SASL_PLAINTEXT
CONNECT_CONSUMER_SSL_ENDPOINT_IDENTIFICATION_ALGORITHM=
CONNECT_CONSUMER_SASL_MECHANISM=PLAIN
CONNECT_GROUP_ID=gcs-kafka-connector
CONNECT_INTERNAL_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_INTERNAL_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter
CONNECT_REST_PORT=28082
CONNECT_PLUGIN_PATH=/usr/share/java,/usr/share/confluent-hub-components
KAFKA_OPTS=-Djava.security.auth.login.config=/usr/share/assets/kafka_admin_account.conf

这是我发现的让 SASL 使用 gcs 连接器所需的所有属性。