diff --git a/downstreamadapter/sink/kafka/sink.go b/downstreamadapter/sink/kafka/sink.go
index 8cd9ed3be3..0aee34e123 100644
--- a/downstreamadapter/sink/kafka/sink.go
+++ b/downstreamadapter/sink/kafka/sink.go
@@ -126,23 +126,24 @@ func Verify(ctx context.Context, changefeedID commonType.ChangeFeedID, uri *url.
 	if err != nil {
 		return errors.Trace(err)
 	}
-	if _, exists := topics[topic]; exists {
-		return nil
-	}
-
-	topicConfig := options.DeriveTopicConfig()
-	if !topicConfig.AutoCreate {
-		return errors.ErrKafkaInvalidConfig.GenWithStack("`auto-create-topic` is false, and %s not found", topic)
-	}
+	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)
+		// 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)
diff --git a/downstreamadapter/sink/kafka/sink_test.go b/downstreamadapter/sink/kafka/sink_test.go
index bfed3536c0..892b3e7c92 100644
--- a/downstreamadapter/sink/kafka/sink_test.go
+++ b/downstreamadapter/sink/kafka/sink_test.go
@@ -16,10 +16,13 @@ package kafka
 import (
 	"context"
 	"fmt"
+	"net/http"
+	"net/http/httptest"
 	"net/url"
 	"testing"
 	"time"
 
+	"github.com/IBM/sarama"
 	"github.com/golang/mock/gomock"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/columnselector"
 	"github.com/pingcap/ticdc/downstreamadapter/sink/eventrouter"
@@ -37,17 +40,41 @@ import (
 
 const kafkaSinkTestTopic = "mock_topic"
 
-func TestVerifyValidatesEncoderConfigBeforeKafkaConnection(t *testing.T) {
-	openProtocol := config.ProtocolOpen.String()
-	sinkConfig := &config.SinkConfig{Protocol: &openProtocol}
-	sinkURI, err := url.Parse("kafka://127.0.0.1:1/" + kafkaSinkTestTopic + "?max-batch-size=0")
+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-existing-topic")
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-	err = Verify(ctx, changefeedID, sinkURI, sinkConfig)
-	require.ErrorContains(t, err, "invalid max-batch-size 0")
+	changefeedID := common.NewChangefeedID4Test("test", "verify-invalid-config")
+	err = Verify(context.Background(), changefeedID, sinkURI, sinkConfig)
+	require.ErrorContains(t, err, "ErrAvroSchemaAPIError")
 }
 
 func newKafkaSinkForTestWithProducers(ctx context.Context,
diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go
index c80c7d65a1..6d70c84b52 100644
--- a/downstreamadapter/sink/topicmanager/kafka_topic_manager.go
+++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager.go
@@ -15,7 +15,6 @@ package topicmanager
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
@@ -239,9 +238,11 @@ func (m *kafkaTopicManager) createTopic(
 	topicName string,
 ) (int32, error) {
 	if !m.cfg.AutoCreate {
-		return 0, errors.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
 	}
 
 	start := time.Now()
diff --git a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
index 521b139c52..45ee322c63 100644
--- a/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
+++ b/downstreamadapter/sink/topicmanager/kafka_topic_manager_test.go
@@ -82,6 +82,7 @@ func TestCreateTopic(t *testing.T) {
 		AutoCreate:        true,
 		PartitionNum:      2,
 		ReplicationFactor: 1,
+		RequiredAcks:      kafka.WaitForAll,
 	}
 
 	changefeedID := common.NewChangefeedID4Test("test", "test")
@@ -137,6 +138,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)
@@ -151,7 +153,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")
@@ -182,6 +189,39 @@ func TestCreateTopic(t *testing.T) {
 	require.False(t, gotFailedTopicValidateOnly)
 }
 
+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) {
 	t.Parallel()
 
diff --git a/pkg/sink/kafka/options.go b/pkg/sink/kafka/options.go
index 301471917c..506e6287ef 100644
--- a/pkg/sink/kafka/options.go
+++ b/pkg/sink/kafka/options.go
@@ -14,7 +14,6 @@
 package kafka
 
 import (
-	"context"
 	"encoding/base64"
 	"fmt"
 	"net/http"
@@ -244,6 +243,9 @@ func (o *options) Apply(changefeedID common.ChangeFeedID,
 	}
 
 	if urlParameter.ReplicationFactor != nil {
+		if *urlParameter.ReplicationFactor <= 0 {
+			return errors.ErrKafkaInvalidConfig.GenWithStack("invalid replication-factor %d", *urlParameter.ReplicationFactor)
+		}
 		o.ReplicationFactor = *urlParameter.ReplicationFactor
 	}
 
@@ -536,11 +538,12 @@ func (o *options) applySASL(urlParameter *urlConfig, sinkConfig *config.SinkConf
 	return nil
 }
 
-// AutoCreateTopicConfig is used to create topic configuration.
+// AutoCreateTopicConfig contains settings used to create and validate a topic.
 type AutoCreateTopicConfig struct {
 	AutoCreate        bool
 	PartitionNum      int32
 	ReplicationFactor int16
+	RequiredAcks      RequiredAcks
 }
 
 func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig {
@@ -548,9 +551,42 @@ func (o *options) DeriveTopicConfig() *AutoCreateTopicConfig {
 		AutoCreate:        o.AutoCreate,
 		PartitionNum:      o.PartitionNum,
 		ReplicationFactor: o.ReplicationFactor,
+		RequiredAcks:      o.RequiredAcks,
 	}
 }
 
+// ValidateReplicationFactor checks whether a topic created with this config
+// can satisfy the configured acknowledgment requirement.
+func (c *AutoCreateTopicConfig) ValidateReplicationFactor(admin ClusterAdminClient) error {
+	if c.RequiredAcks != WaitForAll {
+		return nil
+	}
+
+	raw, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName)
+	if err != nil {
+		log.Warn("cannot get Kafka broker configuration, assume replication factor is valid",
+			zap.String("configName", MinInsyncReplicasConfigName),
+			zap.Int16("replicationFactor", c.ReplicationFactor),
+			zap.Error(err))
+		return nil
+	}
+	minInsyncReplicas, err := strconv.Atoi(raw)
+	if err != nil {
+		return err
+	}
+
+	if int(c.ReplicationFactor) < minInsyncReplicas {
+		return errors.ErrKafkaInvalidConfig.GenWithStack(
+			"TiCDC Kafka sink's `request.required.acks` defaults to -1, "+
+				"TiCDC cannot deliver messages when the `replication-factor` %d "+
+				"is smaller than the `min.insync.replicas` %d of broker",
+			c.ReplicationFactor, minInsyncReplicas,
+		)
+	}
+
+	return nil
+}
+
 var (
 	validClientID     = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`)
 	commonInvalidChar = regexp.MustCompile(`[\?:,"]`)
@@ -578,7 +614,6 @@ func NewKafkaClientID(captureAddr string,
 // It overwrites MaxMessageBytes with the final producer message limit derived
 // from the topic or broker configuration.
 func adjustOptions(
-	ctx context.Context,
 	changefeedID common.ChangeFeedID,
 	admin ClusterAdminClient,
 	options *options,
@@ -589,26 +624,11 @@ func adjustOptions(
 		return errors.Trace(err)
 	}
 
-	if err = validateRequiredAcks(ctx, admin, topics, topic, options); err != nil {
-		return errors.Trace(err)
-	}
-	return adjustTopicOptions(ctx, changefeedID, admin, options, topic, topics)
-}
-
-func adjustTopicOptions(
-	ctx context.Context,
-	changefeedID common.ChangeFeedID,
-	admin ClusterAdminClient,
-	options *options,
-	topic string,
-	topics map[string]TopicDetail,
-) error {
 	info, exists := topics[topic]
 	// once we have found the topic, no matter `auto-create-topic`,
 	// make sure user input parameters are valid.
-	var err error
 	if exists {
-		err = adjustExistingTopicOption(ctx, changefeedID, admin, options, topic, info)
+		err = adjustExistingTopicOption(changefeedID, admin, options, topic, info)
 	} else {
 		adjustNewTopicOptions(admin, changefeedID, options, topic)
 	}
@@ -620,31 +640,14 @@ func adjustTopicOptions(
 	return nil
 }
 
-func validateRequiredAcks(
-	ctx context.Context,
-	admin ClusterAdminClient,
-	topics map[string]TopicDetail,
-	topic string,
-	options *options,
-) error {
-	// Only check replicationFactor >= minInsyncReplicas when producer's required acks is -1.
-	// If we don't check it, the producer probably can not send message to the topic.
-	// Because it will wait for the ack from all replicas. But we do not have enough replicas.
-	if options.RequiredAcks != WaitForAll {
-		return nil
-	}
-	return validateMinInsyncReplicas(ctx, admin, topics, topic, int(options.ReplicationFactor))
-}
-
 func adjustExistingTopicOption(
-	ctx context.Context,
 	changefeedID common.ChangeFeedID,
 	admin ClusterAdminClient,
 	options *options,
 	topic string,
 	info TopicDetail,
 ) error {
-	maxMessageBytes, err := getTopicMaxMessageBytes(ctx, admin, info.Name)
+	maxMessageBytes, err := getTopicMaxMessageBytes(admin, info.Name)
 	if err != nil {
 		log.Warn("`max.message.bytes` not found from topic's configuration, use the option `MaxMessageBytes` as default",
 			zap.String("namespace", changefeedID.Keyspace()), zap.String("changefeed", changefeedID.Name()),
@@ -694,12 +697,11 @@ func adjustNewTopicOptions(
 }
 
 func getTopicMaxMessageBytes(
-	ctx context.Context,
 	admin ClusterAdminClient,
 	topic string,
 ) (int, error) {
 	raw, err := getTopicConfig(
-		ctx, admin, topic,
+		admin, topic,
 		TopicMaxMessageBytesConfigName,
 		BrokerMessageMaxBytesConfigName,
 	)
@@ -725,79 +727,11 @@ func getBrokerMaxMessageBytes(admin ClusterAdminClient) (int, error) {
 	return messageMaxBytes, nil
 }
 
-func validateMinInsyncReplicas(
-	ctx context.Context,
-	admin ClusterAdminClient,
-	topics map[string]TopicDetail,
-	topic string,
-	replicationFactor int,
-) error {
-	minInsyncReplicasConfigGetter := func() (string, bool, error) {
-		info, exists := topics[topic]
-		if exists {
-			minInsyncReplicasStr, err := getTopicConfig(
-				ctx, admin, info.Name,
-				MinInsyncReplicasConfigName,
-				MinInsyncReplicasConfigName)
-			if err != nil {
-				return "", true, err
-			}
-			return minInsyncReplicasStr, true, nil
-		}
-
-		minInsyncReplicasStr, err := admin.GetBrokerConfig(MinInsyncReplicasConfigName)
-		if err != nil {
-			return "", false, err
-		}
-
-		return minInsyncReplicasStr, false, nil
-	}
-
-	minInsyncReplicasStr, exists, err := minInsyncReplicasConfigGetter()
-	if err != nil {
-		// 'min.insync.replica' is invisible to us in Confluent Cloud Kafka.
-		if errors.ErrKafkaConfigNotFound.Equal(err) {
-			log.Warn("TiCDC cannot find `min.insync.replicas` from broker's configuration, " +
-				"please make sure that the replication factor is greater than or equal " +
-				"to the minimum number of in-sync replicas" +
-				"if you want to use `required-acks` = -1." +
-				"Otherwise, TiCDC will not be able to send messages to the topic.")
-		}
-		log.Warn("TiCDC meets error when get `min.insync.replicas` from broker's configuration, assume the config is valid")
-		return nil
-	}
-	minInsyncReplicas, err := strconv.Atoi(minInsyncReplicasStr)
-	if err != nil {
-		return err
-	}
-
-	configFrom := "topic"
-	if !exists {
-		configFrom = "broker"
-	}
-
-	if replicationFactor < minInsyncReplicas {
-		msg := fmt.Sprintf("`replication-factor` cannot be smaller than the `%s` of %s",
-			MinInsyncReplicasConfigName, configFrom)
-		log.Error(msg, zap.Int("replication-factor", replicationFactor),
-			zap.Int("min.insync.replicas", minInsyncReplicas))
-		return errors.ErrKafkaInvalidConfig.GenWithStack(
-			"TiCDC Kafka sink's `request.required.acks` defaults to -1, "+
-				"TiCDC cannot deliver messages when the `replication-factor` %d "+
-				"is smaller than the `min.insync.replicas` %d of %s",
-			replicationFactor, minInsyncReplicas, configFrom,
-		)
-	}
-
-	return nil
-}
-
 // getTopicConfig gets topic config by name.
 // If the topic does not have this configuration,
 // we will try to get it from the broker's configuration.
 // NOTICE: The configuration names of topic and broker may be different for the same configuration.
 func getTopicConfig(
-	ctx context.Context,
 	admin ClusterAdminClient,
 	topicName string,
 	topicConfigName string,
diff --git a/pkg/sink/kafka/options_test.go b/pkg/sink/kafka/options_test.go
index ea5d2e7147..d922166c40 100644
--- a/pkg/sink/kafka/options_test.go
+++ b/pkg/sink/kafka/options_test.go
@@ -205,6 +205,18 @@ func TestCompleteOptions(t *testing.T) {
 	options = NewOptions()
 	err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink)
 	require.Regexp(t, ".*invalid syntax.*", errors.Cause(err))
+	for _, replicationFactor := range []string{"0", "-1"} {
+		uri = "kafka://127.0.0.1:9092/abc?replication-factor=" + replicationFactor
+		sinkURI, err = url.Parse(uri)
+		require.NoError(t, err)
+		options = NewOptions()
+		err = options.Apply(
+			commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"),
+			sinkURI,
+			config.GetDefaultReplicaConfig().Sink,
+		)
+		require.ErrorContains(t, err, "invalid replication-factor "+replicationFactor)
+	}
 
 	// Illegal max-message-bytes.
 	uri = "kafka://127.0.0.1:9092/abc?kafka-version=2.6.0&max-message-bytes=a"
@@ -471,7 +483,7 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t
 			expectedProducerLimit := adminFixture.brokerMessageMaxBytes()
 
 			ctx := context.Background()
-			err = adjustOptions(ctx, changefeedID, adminClient, options, topicName)
+			err = adjustOptions(changefeedID, adminClient, options, topicName)
 			require.NoError(t, err)
 
 			saramaConfig, err := newSaramaConfig(ctx, options)
@@ -489,86 +501,39 @@ func TestAdjustConfigFallsBackToBrokerMessageMaxBytesWhenTopicConfigMissing(t *t
 	}
 }
 
-func TestAdjustConfigMinInsyncReplicas(t *testing.T) {
+func TestValidateReplicationFactor(t *testing.T) {
 	adminFixture := newKafkaAdminFixture(t)
 	adminClient := adminFixture.admin
-
-	options := NewOptions()
-	options.BrokerEndpoints = []string{"127.0.0.1:9092"}
-
-	// Report an error if the replication-factor is less than min.insync.replicas
-	// when the topic does not exist.
 	adminFixture.setMinInsyncReplicas("2")
-	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
 
-	ctx := context.Background()
-	err := adjustOptions(
-		ctx,
-		changefeedID,
-		adminClient,
-		options,
-		"create-new-fail-invalid-min-insync-replicas",
-	)
+	topicConfig := &AutoCreateTopicConfig{
+		AutoCreate:        true,
+		ReplicationFactor: 1,
+		RequiredAcks:      WaitForAll,
+	}
+	err := topicConfig.ValidateReplicationFactor(adminClient)
 	require.Regexp(
 		t,
 		".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of broker.*",
 		errors.Cause(err),
 	)
 
-	// topic not exist, and `min.insync.replicas` not found in broker's configuration
-	adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName)
-	topicName := "no-topic-no-min-insync-replicas"
-	err = adjustOptions(ctx, changefeedID, adminClient, options, "no-topic-no-min-insync-replicas")
-	require.Nil(t, err)
-	err = adminClient.CreateTopic(&TopicDetail{
-		Name:              topicName,
+	localAcksConfig := &AutoCreateTopicConfig{
+		AutoCreate:        true,
 		ReplicationFactor: 1,
-	}, false)
-	require.ErrorIs(t, err, sarama.ErrPolicyViolation)
-
-	// Report an error if the replication-factor is less than min.insync.replicas
-	// when the topic does exist.
-
-	// topic exist, but `min.insync.replicas` not found in topic and broker configuration
-	topicName = "topic-no-options-entry"
-	err = adminClient.CreateTopic(&TopicDetail{
-		Name:              topicName,
-		ReplicationFactor: 3,
-		NumPartitions:     3,
-	}, false)
-	require.Nil(t, err)
-	err = adjustOptions(ctx, changefeedID, adminClient, options, topicName)
-	require.Nil(t, err)
-
-	// topic found, and have `min.insync.replicas`, but set to 2, larger than `replication-factor`.
-	adminFixture.setMinInsyncReplicas("2")
-	err = adjustOptions(ctx, changefeedID, adminClient, options, defaultMockTopicName)
-	require.Regexp(t,
-		".*`replication-factor` 1 is smaller than the `min.insync.replicas` 2 of topic.*",
-		errors.Cause(err),
-	)
-}
-
-func TestSkipAdjustConfigMinInsyncReplicasWhenRequiredAcksIsNotWailAll(t *testing.T) {
-	adminFixture := newKafkaAdminFixture(t)
-	adminClient := adminFixture.admin
-
-	options := NewOptions()
-	options.BrokerEndpoints = []string{"127.0.0.1:9092"}
-	options.RequiredAcks = WaitForLocal
-
-	changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
+		RequiredAcks:      WaitForLocal,
+	}
+	err = localAcksConfig.ValidateReplicationFactor(adminClient)
+	require.NoError(t, err)
 
-	// Do not report an error if the replication-factor is less than min.insync.replicas(1<2).
-	adminFixture.setMinInsyncReplicas("2")
-	err := adjustOptions(
-		context.Background(),
-		changefeedID,
-		adminClient,
-		options,
-		"skip-check-min-insync-replicas",
-	)
-	require.Nil(t, err, "Should not report an error when `required-acks` is not `all`")
+	adminFixture.dropBrokerConfig(MinInsyncReplicasConfigName)
+	missingBrokerConfig := &AutoCreateTopicConfig{
+		AutoCreate:        true,
+		ReplicationFactor: 1,
+		RequiredAcks:      WaitForAll,
+	}
+	err = missingBrokerConfig.ValidateReplicationFactor(adminClient)
+	require.NoError(t, err)
 }
 
 func TestCreateProducerFailed(t *testing.T) {
@@ -738,7 +703,6 @@ func TestConfigurationCombinations(t *testing.T) {
 			sinkURI, err := url.Parse(uri)
 			require.Nil(t, err)
 
-			ctx := context.Background()
 			options := NewOptions()
 			err = options.Apply(commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test"), sinkURI, config.GetDefaultReplicaConfig().Sink)
 			require.Nil(t, err)
@@ -754,7 +718,7 @@ func TestConfigurationCombinations(t *testing.T) {
 			}
 
 			changefeedID := commonType.NewChangefeedID4Test(commonType.DefaultKeyspaceName, "test")
-			err = adjustOptions(ctx, changefeedID, adminClient, options, topic)
+			err = adjustOptions(changefeedID, adminClient, options, topic)
 			require.Nil(t, err)
 			require.Equal(t, sourceMaxMessageBytes, options.MaxMessageBytes)
 			require.Equal(
diff --git a/pkg/sink/kafka/sarama_factory.go b/pkg/sink/kafka/sarama_factory.go
index 39c69c0c0b..3be5fd841a 100644
--- a/pkg/sink/kafka/sarama_factory.go
+++ b/pkg/sink/kafka/sarama_factory.go
@@ -57,7 +57,7 @@ func NewSaramaFactory(
 		admin.Close()
 	}()
 
-	if err = adjustOptions(ctx, changefeedID, admin, o, o.Topic); err != nil {
+	if err = adjustOptions(changefeedID, admin, o, o.Topic); err != nil {
 		return nil, errors.Trace(err)
 	}