googleapis / googleapis/java-pubsub-group-kafka-connector
Support custom Credential Providers
- Dominant language
- Java
- Stars
- 50
- Forks
- 37
- Avg merge
- 1d 19h
- Merged PRs (30d)
- 3
Description
**Is your feature request related to a problem? Please describe.**
At our workplace, we are looking to use this lib to connect some of our Kafka instances to ingest data from deployments on GCP via PubSub. We use a custom timed authentication mechanism for GCP based on Hashicorp Vault and as far as we can see we cannot use a custom piece of code/plugin which would support this. As this is something quite critical and useful at my place and there are more ways of auth which people might use later, we have a proposal for this and are happy to implement and contribute if deemed useful.
**Describe the solution you'd like**
As this depends on `com.google.api.gax` which provides the interface `com.google.api.gax.core.CredentialsProvider` of which we can see the various kinds of uses in [here](https://github.com/googleapis/java-pubsub-group-kafka-connector/blob/main/src/main/java/com/google/pubsub/kafka/common/ConnectorCredentialsProvider.java), we propose the following:
1. Add a new config `public static final String GCP_CREDENTIALS_CLASS_CONFIG` to this [class](https://github.com/googleapis/java-pubsub-group-kafka-connector/blob/main/src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java). This would be the fully qualified name of the class which contains the custom implementation of the the CredentialsProvider.
1. Add a clause to load the custom class [here](https://github.com/googleapis/java-pubsub-group-kafka-connector/blob/main/src/main/java/com/google/pubsub/kafka/common/ConnectorCredentialsProvider.java#L52):
```java
if (!credentialsClass.isEmpty()) {
return ConnectorCredentialsProvider.fromClass(credentialsClass);
}
```
1. Add a new method `fromClass()` to this [class](https://github.com/googleapis/java-pubsub-group-kafka-connector/blob/main/src/main/java/com/google/pubsub/kafka/common/ConnectorCredentialsProvider.java).
```java
public static ConnectorCredentialsProvider fromClass(String credentialsClass) {
try {
final Class klass = Class.forName(credentialsClass);
final var obj = klass.getDeclaredConstructor().newInstance();
if (!(obj instanceof CredentialsProvider)) {
throw new IllegalArgumentException("Supplied class %s is not a CredentialsProvider".formatted(credentialsClass));
}
return new ConnectorCredentialsProvider(() -> ((CredentialsProvider) obj).getCredentials());
} catch (Exception e) {
throw new RuntimeException("Error loading class: " + e);
}
}
```
1. Implement the class for Hashicorp Vault:
```java
import com.google.auth.oauth2.GoogleCredentials;
import com.google.api.gax.core.CredentialsProvider;
class HashicorpVaultCredentialsProvider implements CredentialsProvider {
@Override
public GoogleCredentials getCredentials() {
// do the custom calls to get the token here
final var token = "this is from Hashicorp Vault";
return new GoogleCredentials(token);
}
}
```
We think that this should be generic enough for not only our use case but for others too. If there is interest, we are happy to also implement the Hashicorp Vault implementation in the contribution as that's something quite ubiquitous.
Caveats and questions:
- This relies on the interface `com.google.api.gax.core.CredentialsProvider` which is a transitive dependency of the lib. We are not sure if that's something to be considered as a stable thing to expose to users or should a new interface be created?
**Describe alternatives you've considered**
None as this lib is something we use and would like to keep using with these added features!
**Additional context**
None as of now.
Hopefully this is useful and looking forward to collaborating!
Contributor guide
Research direction
Start with src/main/java/com/google/pubsub/kafka/common/ConnectorUtils.java and ConnectorCredentialsProvider.java, then review how existing credential providers are selected and how CredentialsProvider is exposed. Done means a configured fully qualified class can supply credentials, invalid implementations fail clearly, and the chosen API boundary is covered by tests.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- google-cloud, java, kafka
- Domain
- backend, cloud, stream-processing
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 35/100