1
我使用kafka_2.11-0.10.2.1和google here提供的pubsub連接器。我所關心的只是使用獨立連接器將數據從Kafka主題推送到PubSub。我跟着我應該有所有步驟：Kafka PubSub連接器：Jetty ALPN/NPN未正確配置
- 生產的
cps-kafka-connector.jar - 添加了
cps-sink-connector.properties文件在卡夫卡的config目錄。該文件是這樣的：
name=CPSConnector connector.class=com.google.pubsub.kafka.sink.CloudPubSubSinkConnector tasks.max=10 topics=kafka_topic cps.topic=pubsub_topic cps.project=my_gcp_project_12345
connect-standalone.properties使string轉換器，因爲我的目的是隻發送字符串消息：key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.storage.StringConverter
kafka_topic和發送一些消息如下：$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic kafka_topic $ hello streams $ kafka streams rock
$ bin/connect-standalone.sh config/connect-standalone.properties config/cps-sink-connector.properties
並且意圖是運行：
$ gcloud beta pubsub subscriptions pull subscription_to_pubsub_topic
收集這些信息。但是，下面的錯誤會發生，我無法在他們身上找到我的頭。
他們似乎與jetty-9.2.15.v20160210有關。注意：
[2017-05-04 22:42:26,635] ERROR Commit of WorkerSinkTask{id=CPSConnector-0} offsets threw an unexpected exception:(org.apache.kafka.connect.runtime.WorkerSinkTask:204)
java.lang.RuntimeException: java.util.concurrent.ExecutionException: io.grpc.StatusRuntimeException: UNKNOWN
和
Caused by: java.util.concurrent.ExecutionException: io.grpc.StatusRuntimeException: UNKNOWN ...
Caused by: io.grpc.StatusRuntimeException: UNKNOWN ...
Caused by: java.lang.IllegalArgumentException: Jetty ALPN/NPN has not been properly configured ...
有什麼想法？我如何配置碼頭？我看過一個帖子here其中指出：
沒有標準的Java版本已經內置了支持ALPN今日（！有 跟蹤問題，所以還是給予好評吧），所以我們需要使用碼頭，ALPN （或Jetty-NPN，如果使用Java < 8）OpenJDK的bootclasspath擴展。至 執行此操作，添加一個引用到Jetty alpn-boot jar的路徑的Xbootclasspath JVM選項。
java -Xbootclasspath/p:/path/to/jetty/alpn/extension.jar ...
注意，您必須使用特定於 Java版本正在使用的碼頭 - ALPN罐子的釋放。但是，您可以使用JVM代理程序Jeety-ALPN-Agent爲當前Java版本的 加載正確的Jetty alpn-boot jar文件。爲此，不要添加Xbootclasspath 選項，而是添加一個引用到Jetty alpn-agent jar的路徑的javaagent JVM選項。
java -javaagent:/path/to/jetty-alpn-agent.jar ...
...但我真的不得到如何在我的配置考慮到這一點。有什麼想法嗎？錯誤日誌中的部分如下所示：
...
name = CPSConnector
tasks.max = 10
transforms = null
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig:180)
[2017-05-04 22:42:17,447] INFO TaskConfig values:
task.class = class com.google.pubsub.kafka.sink.CloudPubSubSinkTask
(org.apache.kafka.connect.runtime.TaskConfig:180)
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
[2017-05-04 22:42:17,447] INFO Instantiated task CPSConnector-7 with version 0.10.2.1 of type com.google.pubsub.kafka.sink.CloudPubSubSinkTask (org.apache.kafka.connect.runtime.Worker:317)
[2017-05-04 22:42:17,451] INFO ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [localhost:9092]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = connect-CPSConnector
heartbeat.interval.ms = 3000
interceptor.classes = null
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
(org.apache.kafka.clients.consumer.ConsumerConfig:180)
[2017-05-04 22:42:17,461] INFO Kafka version : 0.10.2.1 (org.apache.kafka.common.utils.AppInfoParser:83)
[2017-05-04 22:42:17,461] INFO Kafka commitId : e89bffd6b2eff799 (org.apache.kafka.common.utils.AppInfoParser:84)
[2017-05-04 22:42:17,462] INFO Creating task CPSConnector-8 (org.apache.kafka.connect.runtime.Worker:305)
[2017-05-04 22:42:17,463] INFO ConnectorConfig values:
connector.class = com.google.pubsub.kafka.sink.CloudPubSubSinkConnector
key.converter = null
name = CPSConnector
tasks.max = 10
transforms = null
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig:180)
[2017-05-04 22:42:17,463] INFO TaskConfig values:
task.class = class com.google.pubsub.kafka.sink.CloudPubSubSinkTask
(org.apache.kafka.connect.runtime.TaskConfig:180)
[2017-05-04 22:42:17,463] INFO Instantiated task CPSConnector-8 with version 0.10.2.1 of type com.google.pubsub.kafka.sink.CloudPubSubSinkTask (org.apache.kafka.connect.runtime.Worker:317)
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
[2017-05-04 22:42:17,465] INFO ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [localhost:9092]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = connect-CPSConnector
heartbeat.interval.ms = 3000
interceptor.classes = null
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
(org.apache.kafka.clients.consumer.ConsumerConfig:180)
[2017-05-04 22:42:17,469] INFO Kafka version : 0.10.2.1 (org.apache.kafka.common.utils.AppInfoParser:83)
[2017-05-04 22:42:17,472] INFO Kafka commitId : e89bffd6b2eff799 (org.apache.kafka.common.utils.AppInfoParser:84)
[2017-05-04 22:42:17,478] INFO Creating task CPSConnector-9 (org.apache.kafka.connect.runtime.Worker:305)
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
[2017-05-04 22:42:17,484] INFO ConnectorConfig values:
connector.class = com.google.pubsub.kafka.sink.CloudPubSubSinkConnector
key.converter = null
name = CPSConnector
tasks.max = 10
transforms = null
value.converter = null
(org.apache.kafka.connect.runtime.ConnectorConfig:180)
[2017-05-04 22:42:17,486] INFO TaskConfig values:
task.class = class com.google.pubsub.kafka.sink.CloudPubSubSinkTask
(org.apache.kafka.connect.runtime.TaskConfig:180)
[2017-05-04 22:42:17,486] INFO Instantiated task CPSConnector-9 with version 0.10.2.1 of type com.google.pubsub.kafka.sink.CloudPubSubSinkTask (org.apache.kafka.connect.runtime.Worker:317)
[2017-05-04 22:42:17,486] INFO ConsumerConfig values:
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [localhost:9092]
check.crcs = true
client.id =
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = connect-CPSConnector
heartbeat.interval.ms = 3000
interceptor.classes = null
key.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.ByteArrayDeserializer
(org.apache.kafka.clients.consumer.ConsumerConfig:180)
[2017-05-04 22:42:17,492] INFO Kafka version : 0.10.2.1 (org.apache.kafka.common.utils.AppInfoParser:83)
[2017-05-04 22:42:17,493] INFO Kafka commitId : e89bffd6b2eff799 (org.apache.kafka.common.utils.AppInfoParser:84)
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
[2017-05-04 22:42:17,496] INFO Created connector CPSConnector (org.apache.kafka.connect.cli.ConnectStandalone:90)
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
May 04, 2017 10:42:17 PM io.grpc.internal.ManagedChannelImpl <init>
INFO: [[email protected]] Created with target pubsub.googleapis.com:443
...
Caused by: io.grpc.StatusRuntimeException: UNKNOWN
at io.grpc.Status.asRuntimeException(Status.java:545)
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:417)
at io.grpc.ClientInterceptors$CheckedForwardingClientCall.start(ClientInterceptors.java:203)
at io.grpc.stub.ClientCalls.startCall(ClientCalls.java:248)
at io.grpc.stub.ClientCalls.asyncUnaryRequestCall(ClientCalls.java:227)
at io.grpc.stub.ClientCalls.futureUnaryCall(ClientCalls.java:186)
at com.google.pubsub.v1.PublisherGrpc$PublisherFutureStub.publish(PublisherGrpc.java:480)
at com.google.pubsub.kafka.sink.CloudPubSubGRPCPublisher.publish(CloudPubSubGRPCPublisher.java:44)
at com.google.pubsub.kafka.sink.CloudPubSubRoundRobinPublisher.publish(CloudPubSubRoundRobinPublisher.java:43)
at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.publishMessagesForPartition(CloudPubSubSinkTask.java:321)
at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:265)
... 22 more
Caused by: java.lang.IllegalArgumentException: Jetty ALPN/NPN has not been properly configured.
at io.grpc.netty.GrpcSslContexts.selectApplicationProtocolConfig(GrpcSslContexts.java:153)
at io.grpc.netty.GrpcSslContexts.configure(GrpcSslContexts.java:130)
at io.grpc.netty.GrpcSslContexts.configure(GrpcSslContexts.java:119)
at io.grpc.netty.GrpcSslContexts.forClient(GrpcSslContexts.java:90)
at io.grpc.netty.NettyChannelBuilder.createProtocolNegotiator(NettyChannelBuilder.java:263)
at io.grpc.netty.NettyChannelBuilder$NettyTransportFactory.newClientTransport(NettyChannelBuilder.java:322)
at io.grpc.internal.CallCredentialsApplyingTransportFactory.newClientTransport(CallCredentialsApplyingTransportFactory.java:62)
at io.grpc.internal.TransportSet.startNewTransport(TransportSet.java:199)
at io.grpc.internal.TransportSet.obtainActiveTransport(TransportSet.java:179)
at io.grpc.internal.ManagedChannelImpl$3.getTransport(ManagedChannelImpl.java:476)
at io.grpc.internal.ManagedChannelImpl$3.getTransport(ManagedChannelImpl.java:432)
at io.grpc.DummyLoadBalancerFactory$DummyLoadBalancer.pickTransport(DummyLoadBalancerFactory.java:105)
at io.grpc.internal.ManagedChannelImpl$1.get(ManagedChannelImpl.java:149)
at io.grpc.internal.ClientCallImpl.start(ClientCallImpl.java:201)
at io.grpc.auth.ClientAuthInterceptor$1.checkedStart(ClientAuthInterceptor.java:104)
at io.grpc.ClientInterceptors$CheckedForwardingClientCall.start(ClientInterceptors.java:195)
... 30 more