kafka: verify replication-factor when need to create the topic (#5715)

downstreamadapter/sink/kafka/sink.go

@@ -68,9 +68,91 @@ func (s *sink) SinkType() commonType.SinkType {
 }
 
 func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.URL, sinkConfig *config.SinkConfig) error {
+<<<<<<< HEAD
 	comp, _, err := newKafkaSinkComponent(ctx, changefeedID, uri, sinkConfig)
 	defer comp.close()
 	return err
+=======
+	protocol, err := helper.GetProtocol(util.GetOrZero(sinkConfig.Protocol))
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	topic, err := helper.GetTopic(uri)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	options := kafka.NewOptions()
+	if err = options.Apply(changefeedID, uri, sinkConfig); err != nil {
+		return errors.WrapError(errors.ErrKafkaInvalidConfig, err)
+	}
+	options.Topic = topic
+
+	encoderConfig, err := helper.GetEncoderConfig(
+		changefeedID, uri, protocol, sinkConfig,
+		options.MaxMessageBytes, options.MaxBatchedBytes,
+	)
+	if err != nil {
+		return errors.Trace(err)
+	}
+
+	claimCheck, err := claimcheck.New(ctx, encoderConfig.LargeMessageHandle, changefeedID)
+	if err != nil {
+		return err
+	}
+	defer claimCheck.Close()
+
+	isAvroLike := protocol == config.ProtocolAvro || protocol == config.ProtocolDebeziumAvro
+	if _, err = eventrouter.NewEventRouter(sinkConfig, topic, false, isAvroLike); err != nil {
+		return errors.Trace(err)
+	}
+
+	if _, err = columnselector.New(sinkConfig); err != nil {
+		return errors.Trace(err)
+	}
+
+	factory, err := kafka.NewSaramaFactory(ctx, options, changefeedID)
+	if err != nil {
+		return errors.WrapError(errors.ErrKafkaNewProducer, err)
+	}
+
+	adminClient, err := factory.AdminClient(ctx)
+	if err != nil {
+		return errors.WrapError(errors.ErrKafkaNewProducer, err)
+	}
+	defer adminClient.Close()
+
+	topics, err := adminClient.GetTopicsMeta([]string{topic}, false)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if _, exists := topics[topic]; !exists {
+		topicConfig := options.DeriveTopicConfig()
+		if !topicConfig.AutoCreate {
+			return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic)
+		}
+		if err = topicConfig.ValidateReplicationFactor(adminClient); err != nil {
+			return err
+		}
+
+		// the topic is not created, only validate.
+		err = adminClient.CreateTopic(&kafka.TopicDetail{
+			Name:              topic,
+			NumPartitions:     topicConfig.PartitionNum,
+			ReplicationFactor: topicConfig.ReplicationFactor,
+		}, true)
+		if err != nil {
+			return errors.WrapError(errors.ErrKafkaCreateTopic, err)
+		}
+	}
+
+	_, err = codec.NewEventEncoder(ctx, encoderConfig, claimCheck)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	return nil
+>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715))
 }
 
 func New(

downstreamadapter/sink/kafka/sink_test.go

@@ -16,11 +16,20 @@
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"testing"
 	"time"
 
+<<<<<<< HEAD
 	"github.com/pingcap/errors"
+=======
+	"github.com/IBM/sarama"
+	"github.com/golang/mock/gomock"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
+	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
+>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715))
 	"github.com/pingcap/ticdc/downstreamadapter/sink/helper"
 	"github.com/pingcap/ticdc/pkg/common"
 	commonEvent "github.com/pingcap/ticdc/pkg/common/event"
@@ -32,6 +41,48 @@
 	"go.uber.org/atomic"
 )
 
+<<<<<<< HEAD
+=======
+const kafkaSinkTestTopic = "mock_topic"
+
+func TestVerifyInvalidConfig(t *testing.T) {
+	broker := sarama.NewMockBroker(t, 1)
+	defer broker.Close()
+	broker.SetHandlerByMap(map[string]sarama.MockResponse{
+		"ApiVersionsRequest": sarama.NewMockApiVersionsResponse(t).SetApiKeys(
+			[]sarama.ApiVersionsResponseKey{
+				{ApiKey: 0},
+				{ApiKey: 1},
+				{ApiKey: 2},
+				{ApiKey: 3, MaxVersion: 9},
+			}),
+		"MetadataRequest": sarama.NewMockMetadataResponse(t).
+			SetController(broker.BrokerID()).
+			SetBroker(broker.Addr(), broker.BrokerID()).
+			SetLeader(kafkaSinkTestTopic, 0, broker.BrokerID()),
+		"DescribeConfigsRequest": sarama.NewMockDescribeConfigsResponse(t),
+	})
+
+	schemaRegistry := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		http.Error(w, "invalid response", http.StatusInternalServerError)
+	}))
+	defer schemaRegistry.Close()
+
+	avroProtocol := config.ProtocolAvro.String()
+	sinkConfig := &config.SinkConfig{
+		Protocol:       &avroProtocol,
+		SchemaRegistry: &schemaRegistry.URL,
+	}
+	sinkURI, err := url.Parse("kafka://" + broker.Addr() + "/" + kafkaSinkTestTopic +
+		"?required-acks=1&kafka-version=2.4.0")
+	require.NoError(t, err)
+
+	changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config")
+	err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig)
+	require.ErrorContains(t, err, "ErrAvroSchemaAPIError")
+}
+
+>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715))
 func newKafkaSinkForTestWithProducers(ctx context.Context,
 	asyncProducer kafka.AsyncProducer,
 	syncProducer kafka.SyncProducer,

downstreamadapter/sink/topicmanager/kafka_topic_manager.go

@@ -15,7 +15,6 @@ package topicmanager
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
@@ -242,9 +241,17 @@ func (m *kafkaTopicManager) createTopic(
 	topicName string,
 ) (int32, error) {
 	if !m.cfg.AutoCreate {
+<<<<<<< HEAD
 		return 0, cerror.ErrKafkaInvalidConfig.GenWithStack(
 			fmt.Sprintf("`auto-create-topic` is false, "+
 				"and %s not found", topicName))
+=======
+		return 0, errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topicName)
+	}
+
+	if err := m.cfg.ValidateReplicationFactor(m.admin); err != nil {
+		return 0, err
+>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715))
 	}
 
 	start := time.Now()

downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go

@@ -31,6 +31,7 @@ func TestCreateTopic(t *testing.T) {
 		AutoCreate:        true,
 		PartitionNum:      2,
 		ReplicationFactor: 1,
+		RequiredAcks:      kafka.WaitForAll,
 	}
 
 	changefeedID := common.NewChangefeedID4Test("test", "test")
@@ -41,6 +42,7 @@ func TestCreateTopic(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
 
+	cfg.RequiredAcks = kafka.WaitForLocal
 	partitionNum, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic")
 	require.NoError(t, err)
 	require.Equal(t, int32(2), partitionNum)
@@ -49,7 +51,12 @@ func TestCreateTopic(t *testing.T) {
 	require.Equal(t, int32(2), partitionsNum)
 
 	// Try to create a topic without auto create.
-	cfg.AutoCreate = false
+	cfg = &kafka.AutoCreateTopicConfig{
+		AutoCreate:        false,
+		PartitionNum:      2,
+		ReplicationFactor: 1,
+		RequiredAcks:      kafka.WaitForAll,
+	}
 	manager = newKafkaTopicManager(ctx, "new-topic2", changefeedID, adminClient, cfg)
 	defer manager.Close()
 	_, err = manager.CreateTopicAndWaitUntilVisible(ctx, "new-topic2")
@@ -77,7 +84,44 @@ func TestCreateTopic(t *testing.T) {
 	)
 }
 
+<<<<<<< HEAD
 func TestCreateTopicWithDelay(t *testing.T) {
+=======
+func TestCreateTopicValidatesReplicationFactor(t *testing.T) {
+	t.Parallel()
+
+	ctrl := gomock.NewController(t)
+	adminClient := kafka.NewMockClusterAdminClient(ctrl)
+	topic := "new-topic"
+	gomock.InOrder(
+		adminClient.EXPECT().GetTopicsMeta([]string{topic}, true).
+			Return(map[string]kafka.TopicDetail{}, nil),
+		adminClient.EXPECT().GetTopicsMeta([]string{topic}, false).
+			Return(map[string]kafka.TopicDetail{}, nil),
+		adminClient.EXPECT().GetBrokerConfig(kafka.MinInsyncReplicasConfigName).
+			Return("2", nil),
+	)
+
+	manager := newKafkaTopicManager(
+		context.Background(),
+		topic,
+		common.NewChangefeedID4Test("test", "test"),
+		adminClient,
+		&kafka.AutoCreateTopicConfig{
+			AutoCreate:        true,
+			PartitionNum:      2,
+			ReplicationFactor: 1,
+			RequiredAcks:      kafka.WaitForAll,
+		},
+	)
+	defer manager.Close()
+
+	_, err := manager.CreateTopicAndWaitUntilVisible(context.Background(), topic)
+	require.ErrorContains(t, err, "`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker")
+}
+
+func TestCreateTopicWaitsUntilVisible(t *testing.T) {
+>>>>>>> 0d4929739 (kafka: verify replication-factor when need to create the topic (#5715))
 	t.Parallel()
 
 	adminClient := kafka.NewClusterAdminClientMockImpl()