googleapis / googleapis/java-pubsub-group-kafka-connector
Core: Configuration errors should fail `CloudPubSubSinkConnector`
Nobody has claimed this yet.
- Dominant language
- Java
- Stars
- 50
- Forks
- 37
- Avg merge
- 1d 19h
- Merged PRs (30d)
- 3
Description
Environment details
- OS type and version: Ubuntu jammy
- Java version: 21
- version(s): https://github.com/googleapis/java-pubsub-group-kafka-connector/releases/tag/v1.3.1
Steps to reproduce
- Run the connector and configure it using invalid configuration. For example invalid project, topic or credentials
Expected: The configuration is validated and I cannot post the connector config unless it is valid
Actual: Any configuration is accepted but it then fails at runtime
Example
{
"connector.class": "com.google.pubsub.kafka.sink.CloudPubSubSinkConnector",
"topics": "Any topic",
"cps.project": "foo",
"max.retries": "1",
"metadata.publish": "true",
"gcp.credentials.json": "path to the secrets",
"name": "MyConnector",
"value.converter.schemas.enable": "false",
"errors.tolerance": "all",
"cps.topic": "any topic"
}
Expected: The Sink Connector fails to start
Actual: The Sink Connector starts but prints PERMISSION_DENIED errors in the logs
Stack trace
13:58:20.030 [task-thread-my-sink-connector-0] ERROR org.apache.kafka.connect.runtime.WorkerSinkTask - WorkerSinkTask{id=my-sink--0} Commit of offsets threw an unexpected exception for sequence number 13: null
java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: User not authorized to perform this action.
at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:352) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
at org.apache.kafka.connect.sink.SinkTask.preCommit(SinkTask.java:139) ~[connect-api-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:445) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerSinkTask.commitOffsets(WorkerSinkTask.java:415) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:246) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:226) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:243) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:298) ~[connect-runtime-4.1.0.jar:?]
at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:254) ~[connect-runtime-4.1.0.jar:?]
at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:572) ~[?:?]
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:317) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1144) ~[?:?]
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:642) ~[?:?]
at java.base/java.lang.Thread.run(Thread.java:1583) [?:?]
Caused by: java.util.concurrent.ExecutionException: com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: User not authorized to perform this action.
at com.google.common.util.concurrent.AbstractFuture.getDoneValue(AbstractFuture.java:596) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
at com.google.common.util.concurrent.AbstractFuture.get(AbstractFuture.java:555) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
at com.google.common.util.concurrent.AbstractFuture$TrustedFuture.get(AbstractFuture.java:111) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
at com.google.common.util.concurrent.ForwardingFuture.get(ForwardingFuture.java:67) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
at com.google.pubsub.kafka.sink.CloudPubSubSinkTask.flush(CloudPubSubSinkTask.java:350) ~[pubsub-group-kafka-connector-1.3.1.jar:1.3.1]
... 13 more
Caused by: com.google.api.gax.rpc.PermissionDeniedException: io.grpc.StatusRuntimeException: PERMISSION_DENIED: User not authorized to perform this action.
Additional context
I think that the issue is here https://github.com/googleapis/java-pubsub-group-kafka-connector/blob/2aa7a182f740c6d33577c0f5a61fe098c86fddc6/src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java#L425-L429
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start in src/main/java/com/google/pubsub/kafka/sink/CloudPubSubSinkTask.java around lines 425-429 and trace how connector configuration reaches task startup. Compare the handling of invalid project, topic, and credentials with the expected startup behavior; done means invalid configurations are rejected before runtime publishing and valid configurations still start normally.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- google-cloud, java
- Domain
- backend, cloud
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100