avro: Add name strategy option for schema registry

cdc/sink/codec/avro.go

@@ -542,13 +542,30 @@ func newAvroEventBatchEncoderBuilder(credential *security.Credential, opts map[s
 		return nil, cerror.ErrPrepareAvroFailed.GenWithStack(`Avro protocol requires parameter "registry"`)
 	}
 
+	topic, ok := opts["topic"]
+	if !ok {
+		topic = ""
+	}
+
+	nameStrategy, ok := opts["nameStrategy"]
+	if !ok {
+		switch nameStrategy {
+		case "topic":
+			nameStrategy = "topic"
+		case "legacy", "":
+			nameStrategy = "legacy"
+		default:
+			return nil, fmt.Errorf("invalid name strategy %s", nameStrategy)
+		}
+	}
+
 	ctx := context.Background()
-	keySchemaManager, err := NewAvroSchemaManager(ctx, credential, registryURI, keySchemaSuffix)
+	keySchemaManager, err := NewAvroSchemaManager(ctx, credential, registryURI, keySchemaSuffix, topic, nameStrategy)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}
 
-	valueSchemaManager, err := NewAvroSchemaManager(ctx, credential, registryURI, valueSchemaSuffix)
+	valueSchemaManager, err := NewAvroSchemaManager(ctx, credential, registryURI, valueSchemaSuffix, topic, nameStrategy)
 	if err != nil {
 		return nil, errors.Trace(err)
 	}

cdc/sink/codec/schema_registry.go

@@ -42,6 +42,8 @@ import (
 type AvroSchemaManager struct {
 	registryURL   string
 	subjectSuffix string
+	topic         string
+	nameStrategy  string
 
 	credential *security.Credential
 
@@ -73,7 +75,7 @@ type lookupResponse struct {
 
 // NewAvroSchemaManager creates a new AvroSchemaManager
 func NewAvroSchemaManager(
-	ctx context.Context, credential *security.Credential, registryURL string, subjectSuffix string,
+	ctx context.Context, credential *security.Credential, registryURL string, subjectSuffix string, topic string, nameStrategy string,
 ) (*AvroSchemaManager, error) {
 	registryURL = strings.TrimRight(registryURL, "/")
 	// Test connectivity to the Schema Registry
@@ -108,6 +110,8 @@ func NewAvroSchemaManager(
 		registryURL:   registryURL,
 		cache:         make(map[string]*schemaCacheEntry, 1),
 		subjectSuffix: subjectSuffix,
+		topic:         topic,
+		nameStrategy:  nameStrategy,
 		credential:    credential,
 	}, nil
 }
@@ -385,6 +389,12 @@ func httpRetry(ctx context.Context, credential *security.Credential, r *http.Req
 		if resp.StatusCode >= 200 && resp.StatusCode < 300 || (resp.StatusCode == 404 && allow404) {
 			break
 		}
+
+		// Avoid retrying "HTTP 409 Conflict"
+		if resp.StatusCode == 409 {
+			return nil, errors.New("HTTP server returned status code 409, indicating conflict")
+		}
+
 		log.Warn("HTTP server returned with error", zap.Int("status", resp.StatusCode))
 		_ = resp.Body.Close()
 
@@ -404,5 +414,8 @@ func httpRetry(ctx context.Context, credential *security.Credential, r *http.Req
 
 func (m *AvroSchemaManager) tableNameToSchemaSubject(tableName model.TableName) string {
 	// We should guarantee unique names for subjects
+	if m.nameStrategy == "topic" {
+		return m.topic + m.subjectSuffix
+	}
 	return tableName.Schema + "_" + tableName.Table + m.subjectSuffix
 }

cdc/sink/mq.go

@@ -391,6 +391,7 @@ func newKafkaSaramaSink(ctx context.Context, sinkURI *url.URL, filter *filter.Fi
 	if topic == "" {
 		return nil, cerror.ErrKafkaInvalidConfig.GenWithStack("no topic is specified in sink-uri")
 	}
+	opts["topic"] = topic
 
 	sProducer, err := kafka.NewKafkaSaramaProducer(ctx, topic, producerConfig, errCh)
 	if err != nil {