kafka-connector: support explicit json credentials

[hnousiainen]

Add support for passing in JSON credentials via connector config key 'gcp.credentials.json'.

[googlebot]

Thanks for your pull request. It looks like this may be your first contribution to a Google open source project (if not, look below for help). Before we can look at your pull request, you'll need to sign a Contributor License Agreement (CLA).

📝 Please visit https://cla.developers.google.com/ to sign.

Once you've signed (or fixed any issues), please reply here (e.g. I signed it!) and we'll verify it.

---

What to do if you already signed the CLA

Individual signers

• It's possible we don't have your GitHub username or you're using a different email address on your commit. Check your existing CLA data and verify that your email is set on your git commits.

Corporate signers

• Your company has a Point of Contact who decides which employees are authorized to participate. Ask your POC to be added to the group of authorized contributors. If you don't know who your Point of Contact is, direct the Google project maintainer to go/cla#troubleshoot (Public version).
• The email used to register you as an authorized contributor must be the email used for the Git commit. Check your existing CLA data and verify that your email is set on your git commits.
• The email used to register you as an authorized contributor must also be attached to your GitHub account.

[hnousiainen]

I signed it!

[kamalaboulhosn]

Thanks for this, it looks good! Unfortunately, it looks like CLA signing did not get associated with your GitHub account. Can you please make sure that the email address used for signing the CLA is the one associated with this pull request? You might also need to make sure that the email address is associated with your GitHub account. It's possible that it could be the difference between a work email address and a personal email address. Thanks!

[hnousiainen]

Hmm, I'm pretty sure my github primary email is the same associated with the commit, and that email address is part of the google group that I used for our corporate CLA registration. Any pointers how I could troubleshoot the issue? Thanks!

[kamalaboulhosn]

Ah if it was corporate instead of individual, then it could take a few days to process. I'll keep an eye on it and approve once it shows up.

[hnousiainen]

I just received an email stating "All parties have completed Google Contributor License Agreement for Aiven."

[googlebot]

CLAs look good, thanks!

[sachinksaini]

Hi,

Does this mean kafka connector doesn't work with credential file set in environment variable ?
I am trying to run kafka connector while having json credential file in GOOGLE_APPLICATION_CREDENTIALS environment variable. However I am getting PERMISSION_DENIED error. Please suggest. Thanks in advance.

exception while starting the connector -

Caused by: io.grpc.StatusRuntimeException: PERMISSION_DENIED: User not authorized to perform this action.
at io.grpc.Status.asRuntimeException(Status.java:526)
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:467)
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:41)
at io.grpc.internal.CensusStatsModule$StatsClientInterceptor$1$1.onClose(CensusStatsModule.java:684)
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:41)
at io.grpc.internal.CensusTracingModule$TracingClientInterceptor$1$1.onClose(CensusTracingModule.java:391)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:475)
at io.grpc.internal.ClientCallImpl.access$300(ClientCallImpl.java:63)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:557)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$600(ClientCallImpl.java:478)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:590)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

[kamalaboulhosn]

The connector should fall back to GOOGLE_APPLICATION_CREDENTIALS if it is set and this property is not set in the config properties. Are you trying using a source connector or a sink connector? You might want to verify that you are able to publish to Cloud Pub/Sub directly (if using a sink connector) or subscribe directly (if you are using a source connector) with the environment variable set. If you cannot, it might not be a problem related to the connector, but rather to the credentials or the permissions.

[sachinksaini]

thanks for the quick response. I am using source connector. I believe in my case, it looks like kafka connector is able to read the credentials file, but fails at pubsub. I am able to use the same credential file using gRPC java connection to pull messages and working fine but I would like to use the kafka source connector and failing with above error. Please help.

[kamalaboulhosn]

How are you setting the environment variable and starting up connect? #158 has some information regarding the correct way to do this in order to ensure the environment variable is picked up by Kafka Connect.

[sachinksaini]

I am trying running the kafka source connector on windows 10 to test and environment variable looks correct.
Variable Name: GOOGLE_APPLICATION_CREDENTIALS
Variable value: C:\connector\my-credentials.json

I also tried passing the credential file path in connector properties file as well like as follows :
gcp.credentials.file.path=C:/connector/my-credentials.json

but no luck yet and getting same error -

Caused by: io.grpc.StatusRuntimeException: PERMISSION_DENIED: User not authorized to perform this action.
at io.grpc.Status.asRuntimeException(Status.java:526)
at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:467)
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:41)
at io.grpc.internal.CensusStatsModule$StatsClientInterceptor$1$1.onClose(CensusStatsModule.java:684)
at io.grpc.ForwardingClientCallListener.onClose(ForwardingClientCallListener.java:41)