在没有启用sasl或ssl的情况下，我已经能够成功地连接gcs连接器。当我启用sasl和ssl时；connect standalone似乎无法与代理进行通信。
问题似乎出在gcs接收器许可证管理器上。这是我从日志中发现的，但它们对我真正找出问题所在并没有太大帮助。。。。
日志

[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.属性

# 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=

dockerfile文件

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编写文件

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