BE: SR: Add compatibility w/ GCP SR

[magnusdriver]

What changes did you make? (Give an overview)

Added new cluster property "schemaRegistryAuth.bearerAuthCustomProviderClass" to enable compatibility with new GCP Schema Registries.
This property can be used in config.yaml file like this:

kafka:
  clusters:
    - name: local  # Unique name identifier for the Kafka cluster
       bootstrap-servers: kafka1:9092,kafka2:9092  # List of Kafka broker addresses

       schemaRegistry: https://managedkafka.googleapis.com/v1/projects/PROJECT_ID/locations/REGION/schemaRegistries/REGISTRY_ID
       schema-registry-auth:
         bearer-auth-custom-provider-class: com.google.cloud.hosted.kafka.auth.GcpBearerAuthCredentialProvider

Or can be enabled with the env variable: KAFKA_CLUSTERS_0_SCHEMA_REGISTRY_AUTH_BEARER_AUTH_CUSTOM_PROVIDER_CLASS="com.google.cloud.hosted.kafka.auth.GcpBearerAuthCredentialProvider"

or KAFKA_CLUSTERS_0_SCHEMAREGISTRYAUTH_BEARERAUTHCUSTOMPROVIDERCLASS is also valid.

Important changes in code are:

• Add bearer token generation for auth with GCP Schema Registry in WebClientConfigurator
• Add custom bearer token provider in SchemaRegistrySerde
• Add the new bearerAuthCustomProviderClass config property in kafbat-ui-api
• Add new schema fields kafka-sr-api for compatibility level as GCP Schema Registry return different format.
• Add condition to use compatibility or compatiblityLevel field based on bearerAuthCustomProviderClass
  property in SchemaRegistryService

It's needed to use gcloud auth application-default login to get the credentials to connect to GCP Schema
Registries or run kafka-ui in a compute engine instance or GKE with a service account with the needed permissions.

Note: This functionality only works now with Avro schema types as these are the ones I work with.
It could work with Protobuf schemas, but not with JSON ones as GCP Schema Registries don't
support them.

How Has This Been Tested? (put an "x" (case-sensitive!) next to an item)

• Manually (please, describe, if necessary)

  Changes tested connecting to GCP and Confluent Schema Registries to check:

  • All Schemas registered are shown in Schema Registry dashboard
  • Topic events are deserialized correctly in the topics dashboards.

  It was impossible to pass unit tests even before adding any change :'(

Checklist (put an "x" (case-sensitive!) next to all the items, otherwise the build will fail)

• I have performed a self-review of my own code
• I have commented my code, particularly in hard-to-understand areas
• I have made corresponding changes to the documentation (e.g. ENVIRONMENT VARIABLES)
• My changes generate no new warnings (e.g. Sonar is happy)
• I have added tests that prove my fix is effective or that my feature works
• New and existing unit tests pass locally with my changes

Check out Contributing and Code of Conduct

[github-actions]

Hi magnusdriver! 👋

Welcome, and thank you for opening your first PR in the repo!

Please wait for triaging by our maintainers.

Please take a look at our contributing guide.

[germanosin]

Thanks for your contribution! I believe it would be more effective to implement this in a more generic manner by exposing the BEARER_SCOPE and BEARER_AUTH_CUSTOM_PROVIDER_CLASS parameters through SchemaRegistryAuth. Don't you mind to change it?

[magnusdriver]

I finally exposed the BEARER_AUTH_CUSTOM_PROVIDER_CLASS parameter. I didn't need the bearer scope for this use case. If you think it's still interesting to expose that parameter I'll add it. Maybe pass it to WebClientConfigurator.configureBearerTokenAuth as parameter for future use cases?

[Haarolean]

Left a few inline comments. Once we decide on a higher level of abstractions, we can dive deep into reviewing the concrete implementations. Right now, it's a nailed-down solution not taking future maintainability (or possible expansion towards other implementations) into account.

[Haarolean]

Why has compatibility become no longer required?

[magnusdriver]

I'm trying to change it

[magnusdriver]

I finally added a new compatibilityConfigGcp schema detached from the existent compatibilityConfig. And I had to modify code in SchemaRegistryService class.

If this new modification is not ok then please let me know if you have any recommendation about how to address this as the only alternative I can think of is to add a new kafka-gcp-sr-api and a new GcpSchemaRegistryService class.

[Haarolean]

Not sure it's the best solution here:

• this breaks backward compatibility
• every other new value (AwsCompatibility, AzureCompatibility, etc) will expand this list and break the backward compatibility further

@germanosin what do you think?

[Haarolean]

Why is this called a class? The implementation rather retrieves a token via custom implementation from GCP which is later used as a bearer token value, there are no "classes" in use per se.

[magnusdriver]

I use this to reference this class. And I changed the bearer token implementation for the WebClient using this class too. Sorry 🙏 .

[Haarolean]

Not sure it's the best solution here:

• this breaks backward compatibility
• every other new value (AwsCompatibility, AzureCompatibility, etc) will expand this list and break the backward compatibility further

@germanosin what do you think?

[germanosin]

This approach seems messy. If GCP SR uses a different API, we should define a separate contract for it.

[magnusdriver]

Sorry for taking so long. But I finally managed to decouple both APIs. To make it work I decided to create a GcpSchemaRegistryService and a GcpKafkaSrMapper and make use of them in the SchemaController based on a KafkaCluster variable called "isGcpSchemaRegistry" (bool) initialized in the KafkaClusterFactory. I think with this approach would be easy to undo this changes if Google decides at some point to make their SchemaRegistry fully compliant with the standard SR API.

The Serdes remains as it was because the problematic "compatibility" field is not used there. For that component just the custom auth class is needed.