Merge branch 'main' into main

Author: yaron2

bindings/kafka/metadata.yaml

@@ -352,6 +352,13 @@ metadata:
       It allows sending headers with special characters that are usually not allowed in HTTP headers.
     example: "true"
     default: "false"
+  - name: useAvroJSON
+    type: bool
+    required: false
+    description: |
+      Enables Avro JSON schema for serialization. Only applicable when the subscription uses valueSchemaType=Avro
+    example: "true"
+    default: "false"
   - name: compression
     type: string
     required: false

common/authentication/aws/client_fake.go

@@ -43,12 +43,18 @@ func (m *MockParameterStore) DescribeParametersWithContext(ctx context.Context,
 type MockSecretManager struct {
 	GetSecretValueFn func(context.Context, *secretsmanager.GetSecretValueInput, ...request.Option) (*secretsmanager.GetSecretValueOutput, error)
 	secretsmanageriface.SecretsManagerAPI
+
+	ListSecretsFn func(context.Context, *secretsmanager.ListSecretsInput, ...request.Option) (*secretsmanager.ListSecretsOutput, error)
 }
 
 func (m *MockSecretManager) GetSecretValueWithContext(ctx context.Context, input *secretsmanager.GetSecretValueInput, option ...request.Option) (*secretsmanager.GetSecretValueOutput, error) {
 	return m.GetSecretValueFn(ctx, input, option...)
 }
 
+func (m *MockSecretManager) ListSecretsWithContext(ctx context.Context, input *secretsmanager.ListSecretsInput, option ...request.Option) (*secretsmanager.ListSecretsOutput, error) {
+	return m.ListSecretsFn(ctx, input, option...)
+}
+
 type MockDynamoDB struct {
 	GetItemWithContextFn            func(ctx context.Context, input *dynamodb.GetItemInput, op ...request.Option) (*dynamodb.GetItemOutput, error)
 	PutItemWithContextFn            func(ctx context.Context, input *dynamodb.PutItemInput, op ...request.Option) (*dynamodb.PutItemOutput, error)

common/component/kafka/kafka.go

@@ -223,10 +223,10 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
 	}
 	k.consumeRetryEnabled = meta.ConsumeRetryEnabled
 	k.consumeRetryInterval = meta.ConsumeRetryInterval
-
 	if meta.SchemaRegistryURL != "" {
 		k.logger.Infof("Schema registry URL '%s' provided. Configuring the Schema Registry client.", meta.SchemaRegistryURL)
 		k.srClient = srclient.CreateSchemaRegistryClient(meta.SchemaRegistryURL)
+		k.srClient.CodecJsonEnabled(!meta.UseAvroJSON)
 		// Empty password is a possibility
 		if meta.SchemaRegistryAPIKey != "" {
 			k.srClient.SetCredentials(meta.SchemaRegistryAPIKey, meta.SchemaRegistryAPISecret)
@@ -364,12 +364,7 @@ func (k *Kafka) DeserializeValue(message *sarama.ConsumerMessage, config Subscri
 		if err != nil {
 			return nil, err
 		}
-		// The data coming through is standard JSON. The version currently supported by srclient doesn't support this yet
-		// Use this specific codec instead.
-		codec, err := goavro.NewCodecForStandardJSONFull(schema.Schema())
-		if err != nil {
-			return nil, err
-		}
+		codec := schema.Codec() // The value returned in Avro JSON format
 		native, _, err := codec.NativeFromBinary(message.Value[5:])
 		if err != nil {
 			return nil, err
@@ -405,12 +400,8 @@ func (k *Kafka) getLatestSchema(topic string) (*srclient.Schema, *goavro.Codec,
 		if errSchema != nil {
 			return nil, nil, errSchema
 		}
-		// New JSON standard serialization/Deserialization is not integrated in srclient yet.
-		// Since standard json is passed from dapr, it is needed.
-		codec, errCodec := goavro.NewCodecForStandardJSONFull(schema.Schema())
-		if errCodec != nil {
-			return nil, nil, errCodec
-		}
+		codec := schema.Codec()
+
 		k.latestSchemaCacheWriteLock.Lock()
 		k.latestSchemaCache[subject] = SchemaCacheEntry{schema: schema, codec: codec, expirationTime: time.Now().Add(k.latestSchemaCacheTTL)}
 		k.latestSchemaCacheWriteLock.Unlock()
@@ -420,12 +411,7 @@ func (k *Kafka) getLatestSchema(topic string) (*srclient.Schema, *goavro.Codec,
 	if err != nil {
 		return nil, nil, err
 	}
-	codec, err := goavro.NewCodecForStandardJSONFull(schema.Schema())
-	if err != nil {
-		return nil, nil, err
-	}
-
-	return schema, codec, nil
+	return schema, schema.Codec(), nil
 }
 
 func (k *Kafka) getSchemaRegistyClient() (srclient.ISchemaRegistryClient, error) {