package changefeed

import (
	"fmt"
	"net/url"
	"strings"
	"time"

	"github.com/elliotchance/pie/v2"
	. "github.com/tidbcloud/dataflow-service-e2e/cases"
	"github.com/tidbcloud/dataflow-service-e2e/cases/dedicated/common"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/config"
	"github.com/tidbcloud/dataflow-service-e2e/internal/misc"
	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/cdcapi"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/client/connector_service"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
	ticdc_v1 "github.com/tidbcloud/infra-provider/api/tidbcloud/ticdc-service/v1"
)

type kafkaChangefeedKernelConfigurationCase struct {
	BaseCase
	cdcApiWrapper *cdcapi.CDCClientWrapper
	cfValidator   *common.ChangfeedValidator
	cloudProvider CloudProviderType
}

func NewKafkaChangefeedKernelConfigurationCase(cloudProvider CloudProviderType) E2ECase {
	return &kafkaChangefeedKernelConfigurationCase{
		cloudProvider: cloudProvider,
	}
}

func (k *kafkaChangefeedKernelConfigurationCase) Name() string {
	return fmt.Sprintf("%s%s", k.cloudProvider, "kafkaChangefeedKernelConfiguration")
}

func (k *kafkaChangefeedKernelConfigurationCase) DeclareDependencyRequirements() map[string]DependencyRequirement {
	return map[string]DependencyRequirement{
		"tidb": &TiDBDedicatedRequirement{
			Dedicated: false,
			TiDBDedicatedDimensions: TiDBDedicatedDimensions{
				CloudProvider: k.cloudProvider,
				Region:        GlobalConfig.DefaultRegionOf(k.cloudProvider),
			},
		},
		"kafka": &KafkaRequirement{
			KafkaDimensions: KafkaDimensions{
				NetworkType: DataflowNetworkTypeNetworkTypePublic,
				AuthType:    DataflowKafkaAuthTypeSASLPLAIN,
				TLSEnabled:  false,
			},
		},
	}
}

func (k *kafkaChangefeedKernelConfigurationCase) Run() {
	k.Logger().Info("run %s", k.Name())

	// declare case related vars
	tidbCluster := k.TiDBDedicated("tidb")
	downstream := k.Kafka("kafka")
	changefeedName := k.Name()
	changefeedId := ""
	// init reusable api client and validator
	cdcApiWrapper := cdcapi.NewCDCClientWrapper(k.Logger())
	k.cdcApiWrapper = cdcApiWrapper
	k.cfValidator = common.NewChangfeedValidator(k.Logger())

	changefeedSpec := k.genAddChangefeedSpec(changefeedName, "2rcu", downstream)
	editSpec := connector_service.EditConnectorBody{
		ConnectorType: changefeedSpec.ConnectorType,
		KafkaInfo:     changefeedSpec.KafkaInfo,
		Name:          changefeedName,
	}

	// cleanup changefeeds if exists
	defer func() {
		k.
			RetryDo("delete changefeeds", func() error {
				return k.cfValidator.DeleteChangefeedByName(tidbCluster, changefeedName)
			}, 3, time.Minute).
			Expect("changefeeds not in list", func() error {
				return k.cfValidator.AllChangefeedsNotInListByNames(tidbCluster, changefeedName)
			}, 3, 10*time.Second)
	}()

	// create changefeed
	k.
		RetryDo(fmt.Sprintf("create changefeed %s", changefeedName), func() (err error) {
			resp, err := k.cdcApiWrapper.CreateChangefeed(tidbCluster, changefeedSpec)
			if err != nil {
				return err
			}
			changefeedId = *resp.Payload.ConnectorID
			editSpec.ConnectorID = resp.Payload.ConnectorID
			return nil
		}, 3, time.Minute).
		Expect("changefeeds in list", func() error {
			return k.cfValidator.AreChangefeedsInList(tidbCluster, changefeedId)
		}, 3, 10*time.Second).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 20, time.Minute).
		Expect("compare changefeed configuration", func() error {
			cfg, err := k.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedId)
			if err != nil {
				return err
			}
			return k.compareKafkaChangefeedConfig(changefeedSpec.KafkaInfo, cfg)
		}, 5, time.Minute).
		RetryDo(fmt.Sprintf("pause changefeed %s", changefeedId), func() error {
			return cdcApiWrapper.PauseChangefeed(tidbCluster, changefeedId)
		}, 3, time.Minute).
		RetryDo(fmt.Sprintf("edit changefeed %s", changefeedId), func() error {
			editSpec.KafkaInfo = k.genEditChangefeedSpec(downstream)
			return cdcApiWrapper.EditChangefeed(tidbCluster, changefeedId, editSpec)
		}, 3, time.Minute).
		Expect("compare changefeed configuration", func() error {
			cfg, err := k.cdcApiWrapper.GetChangefeedKernelConfig(tidbCluster, changefeedId)
			if err != nil {
				return err
			}
			return k.compareKafkaChangefeedConfig(editSpec.KafkaInfo, cfg)
		}, 5, time.Minute).
		RetryDo(fmt.Sprintf("resume changefeed %s", changefeedId), func() error {
			return cdcApiWrapper.ResumeChangefeed(tidbCluster, changefeedId)
		}, 3, time.Minute).
		Expect("changefeeeds are running", func() error {
			return k.cfValidator.AreChangefeedsRunning(tidbCluster, changefeedId)
		}, 3, time.Minute)
}

func (k *kafkaChangefeedKernelConfigurationCase) genAddChangefeedSpec(
	name, instanceType string,
	downstream *KafkaEntity,
) connector_service.CreateConnectorBody {
	changefeedSpec := connector_service.CreateConnectorBody{
		ConnectorType: model.NewDataflowConnectorType(model.DataflowConnectorTypeKAFKA),
	}
	changefeedSpec.Name = name
	changefeedSpec.ChangefeedInstanceType = instanceType
	changefeedSpec.NetworkInfo = &model.DataflowNetworkInfo{
		NetworkType: model.NewDataflowNetworkType(model.DataflowNetworkTypeNetworkTypePublic),
	}
	start := time.Now()
	startTso := misc.GoTimeToTSO(start)
	changefeedSpec.KafkaInfo = &model.DataflowKafkaInfo{
		Authentication: &model.DataflowAuthentication{
			AuthType: model.DataflowKafkaAuthType(downstream.AuthType).Pointer(),
			UserName: downstream.User,
			Password: downstream.Password,
		},
		Broker: &model.DataflowBroker{
			BrokerEndpoints: downstream.BootstrapEndpoints,
			Compression:     model.NewDataflowCompressionType(model.DataflowCompressionTypeGZIP),
			KafkaVersion:    model.NewDataflowKafkaVersion(model.DataflowKafkaVersionKAFKAVERSION3XX),
			TLSEnable:       false,
		},
		ColumnSelectors: []*model.DataflowColumnSelector{
			{
				Columns: []string{"abc", "def"},
				Matcher: []string{
					"test.a", "test.b"},
			},
			{
				Columns: []string{"abc"},
				Matcher: []string{
					"*.*"},
			},
		},
		DataFormat: &model.DataflowDataFormat{
			Protocol: model.NewDataflowProtocol(model.DataflowProtocolDEBEZIUM),
			DebeziumConfig: &model.DataflowDebeziumConfig{
				DisableSchema:  false,
				OutputOldValue: true,
			},
		},
		StartPosition: &model.DataflowStartPosition{
			Mode: model.NewDataflowStartMode(model.DataflowStartModeFROMNOW),
			Tso:  fmt.Sprintf("%d", startTso),
		},
		TableConfig: &model.DataflowTableConfig{
			EventFilterRule: []*model.DataflowEventFilterRule{
				{
					IgnoreEvent: []string{"all dml", "all ddl", "update"},
					Matcher:     []string{"a.b", "c.d"},
				},
				{
					IgnoreEvent: []string{"insert", "update"},
					Matcher:     []string{"e.f", "h.j"},
				},
			},
			FilterRule: []string{"test.a", "test.b"},
			Mode:       model.NewDataflowTableMode(model.DataflowTableModeFORCESYNC),
		},
		TopicPartitionConfig: &model.DataflowTopicPartitionConfig{
			DefaultTopic: "test2",
			DispatchType: model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEONETOPIC),
			PartitionDispatchers: []*model.DataflowPartitionDispatcherConfig{
				{
					Columns: []string{
						"abc",
					},
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherCOLUMNS),
				},
				{
					IndexName:     "idx",
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherINDEXVALUE),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherTS),
				},
				{
					Matcher:       []string{"a.b"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherTABLE),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc",
			TopicPrefix:       "efg",
			TopicSuffix:       "hij",
		},
	}
	return changefeedSpec
}

func (k *kafkaChangefeedKernelConfigurationCase) genEditChangefeedSpec(
	downstream *KafkaEntity,
) *model.DataflowKafkaInfo {
	return &model.DataflowKafkaInfo{
		StartPosition: &model.DataflowStartPosition{},
		Authentication: &model.DataflowAuthentication{
			AuthType: model.DataflowKafkaAuthType(downstream.AuthType).Pointer(),
			UserName: downstream.User,
			Password: downstream.Password,
		},
		Broker: &model.DataflowBroker{
			BrokerEndpoints: downstream.BootstrapEndpoints,
			Compression:     model.NewDataflowCompressionType(model.DataflowCompressionTypeLZ4),
			KafkaVersion:    model.NewDataflowKafkaVersion(model.DataflowKafkaVersionKAFKAVERSION2XX),
			TLSEnable:       false,
		},
		ColumnSelectors: []*model.DataflowColumnSelector{
			{
				Columns: []string{"xabc", "xdef"},
				Matcher: []string{
					"xtest.a", "xtest.b"},
			},
			{
				Columns: []string{"xabc"},
				Matcher: []string{
					"*.x"},
			},
		},
		DataFormat: &model.DataflowDataFormat{
			Protocol:            model.NewDataflowProtocol(model.DataflowProtocolCANALJSON),
			EnableTidbExtension: true,
		},
		TableConfig: &model.DataflowTableConfig{
			EventFilterRule: []*model.DataflowEventFilterRule{
				{
					IgnoreEvent: []string{"insert", "delete"},
					Matcher:     []string{"a.f", "e.c"},
				},
				{
					IgnoreEvent: []string{"insert", "update"},
					Matcher:     []string{"xe.f", "xh.j"},
				},
			},
			FilterRule: []string{"xtest.a", "xtest.b"},
			Mode:       model.NewDataflowTableMode(model.DataflowTableModeIGNORENOTSUPPORTTABLE),
		},
		TopicPartitionConfig: &model.DataflowTopicPartitionConfig{
			DefaultTopic: "test5",
			DispatchType: model.NewDataflowDispatchType(model.DataflowDispatchTypeDISPATCHTYPEBYDATABASE),
			PartitionDispatchers: []*model.DataflowPartitionDispatcherConfig{
				{
					Columns: []string{
						"xabc",
					},
					Matcher:       []string{"a.xb"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherCOLUMNS),
				},
				{
					IndexName:     "idx",
					Matcher:       []string{"a.xb"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherINDEXVALUE),
				},
				{
					Matcher:       []string{"a.bb"},
					PartitionType: model.NewDataflowPartitionDispatcher(model.DataflowPartitionDispatcherTABLE),
				},
			},
			PartitionNum:      1,
			ReplicationFactor: 1,
			Separator:         "abc3",
			TopicPrefix:       "efg4",
			TopicSuffix:       "hij2",
		},
	}
}

func (k *kafkaChangefeedKernelConfigurationCase) compareKafkaChangefeedConfig(kafkaCfg *model.DataflowKafkaInfo,
	kernelCfg *ticdc_v1.Changefeed) error {
	sinkURI, err := url.Parse(kernelCfg.SinkUri)
	if err != nil {
		return err
	}
	k.Logger().Info("sinkURI:%s", sinkURI.String())

	if kafkaCfg.Authentication != nil {
		authInfo := kafkaCfg.Authentication
		switch *authInfo.AuthType {
		case model.DataflowKafkaAuthTypeSASLPLAIN:
			if strings.ToLower(sinkURI.Query().Get("sasl-mechanism")) != "plain" {
				return fmt.Errorf("sasl mechanism not expected got %s",
					sinkURI.Query().Get("sasl-password"))
			}
			if authInfo.UserName != sinkURI.Query().Get("sasl-user") {
				return fmt.Errorf("sasl user not match, expected %s, got %s", authInfo.UserName,
					sinkURI.Query().Get("sasl-password"))
			}
			_, ok := sinkURI.Query()["sasl-password"]
			if !ok {
				return fmt.Errorf("sasl password not found")
			}
		}
	}
	if kafkaCfg.Broker != nil {
		broker := kafkaCfg.Broker
		if broker.BrokerEndpoints != sinkURI.Host {
			return fmt.Errorf("broker endpoints not match, expected %s, got %s",
				broker.BrokerEndpoints, sinkURI.Host)
		}
		if strings.ToLower(string(*broker.Compression)) != strings.ToLower(sinkURI.Query().Get("compression")) {
			return fmt.Errorf("compression not match, expected %s, got %s",
				*broker.Compression, sinkURI.Query().Get("compression"))
		}
		if strings.ToLower(transferKafkaVersion(*broker.KafkaVersion)) != strings.ToLower(sinkURI.Query().Get("kafka-version")) {
			return fmt.Errorf("kafka version not match, expected %s, got %s",
				*broker.KafkaVersion, sinkURI.Query().Get("kafka-version"))
		}
		if broker.TLSEnable != (sinkURI.Query().Get("enable-tls") == "true") {
			return fmt.Errorf("tls enable not match, expected %v, got %v",
				broker.TLSEnable, sinkURI.Query().Get("enable-tls") == "true")
		}
	}
	if kafkaCfg.DataFormat != nil {
		dataFormat := kafkaCfg.DataFormat
		if strings.ToLower(transferProtocol(*dataFormat.Protocol)) != strings.ToLower(sinkURI.Query().Get("protocol")) {
			return fmt.Errorf("protocol not match, expected %s, got %s",
				*dataFormat.Protocol, sinkURI.Query().Get("protocol"))
		}
		if dataFormat.DebeziumConfig != nil {
			debeziumCfg := dataFormat.DebeziumConfig
			if kernelCfg.Config.Sink.DebeziumDisableSchema != nil && debeziumCfg.DisableSchema != *kernelCfg.Config.Sink.DebeziumDisableSchema {
				return fmt.Errorf("disable debezium schema not match, expected %v, got %v",
					debeziumCfg.DisableSchema, kernelCfg.Config.Sink.DebeziumDisableSchema)
			}
			if kernelCfg.Config.Sink.Debezium.OutputOldValue != nil && debeziumCfg.OutputOldValue != *kernelCfg.Config.Sink.Debezium.OutputOldValue {
				return fmt.Errorf("output old value not match, expected %v, got %v",
					debeziumCfg.OutputOldValue, kernelCfg.Config.Sink.Debezium.OutputOldValue)
			}
		}
		v, ok := sinkURI.Query()["enable-tidb-extension"]
		if ok && dataFormat.EnableTidbExtension != (v[0] == "true") {
			return fmt.Errorf("enable tidb extension not match, expected %v, got %v",
				dataFormat.EnableTidbExtension, v[0] == "true")

		}
		v, ok = sinkURI.Query()["avro-bigint-unsigned-handling-mode"]
		if ok && string(*dataFormat.BigintUnsignedHandlingMode) != v[0] {
			return fmt.Errorf("bigint unsigned handling mode not match, expected %s, got %s",
				*dataFormat.BigintUnsignedHandlingMode, v[0])
		}
		v, ok = sinkURI.Query()["avro-decimal-handling-mode"]
		if ok && string(*dataFormat.DecimalHandlingMode) != v[0] {
			return fmt.Errorf("decimal handling mode not match, expected %s, got %s",
				*dataFormat.DecimalHandlingMode, v[0])
		}
	}
	if kafkaCfg.ColumnSelectors != nil {
		if len(kafkaCfg.ColumnSelectors) != len(kernelCfg.Config.Sink.ColumnSelectors) {
			return fmt.Errorf("column selectors length not match, expected %d, got %d",
				len(kafkaCfg.ColumnSelectors), len(kernelCfg.Config.Sink.ColumnSelectors))
		}
		for i, apiSelector := range kafkaCfg.ColumnSelectors {
			kernelSelector := kernelCfg.Config.Sink.ColumnSelectors[i]
			if !pie.Equals(apiSelector.Columns, kernelSelector.Columns) {
				return fmt.Errorf("column selectors column not match, expected %v, got %v",
					apiSelector.Matcher, kernelSelector.Matcher)
			}
			if !pie.Equals(apiSelector.Matcher, kernelSelector.Matcher) {
				return fmt.Errorf("column selectors matcher not match, expected %v, got %v",
					apiSelector.Matcher, kernelSelector.Matcher)
			}
		}
	}
	if kafkaCfg.TableConfig != nil {
		tableCfg := kafkaCfg.TableConfig
		if len(tableCfg.EventFilterRule) != len(kernelCfg.Config.Filter.Rules) {
			return fmt.Errorf("event filter rules length not match, expected %d, got %d",
				len(tableCfg.EventFilterRule), len(kernelCfg.Config.Filter.Rules))
		}
		for i, apiRule := range tableCfg.EventFilterRule {
			kernelRule := kernelCfg.Config.Filter.EventFilters[i]
			if !pie.Equals(apiRule.IgnoreEvent, kernelRule.IgnoreEvent) {
				return fmt.Errorf("event filter rules ignore event not match, expected %v, got %v",
					apiRule.IgnoreEvent, kernelRule.IgnoreEvent)
			}
			if !pie.Equals(apiRule.Matcher, kernelRule.Matcher) {
				return fmt.Errorf("event filter rules ignore event not match, expected %v, got %v",
					apiRule.IgnoreEvent, kernelRule.IgnoreEvent)
			}
		}
		if *tableCfg.Mode == model.DataflowTableModeFORCESYNC &&
			(kernelCfg.Config.ForceReplicate != nil && !*kernelCfg.Config.ForceReplicate) {
			return fmt.Errorf("force replicate not match, expected %v, got %v",
				true, kernelCfg.Config.ForceReplicate)
		}
		if *tableCfg.Mode == model.DataflowTableModeIGNORENOTSUPPORTTABLE &&
			(kernelCfg.Config.IgnoreIneligibleTable != nil && !*kernelCfg.Config.IgnoreIneligibleTable) {
			return fmt.Errorf("ignore ineligible table not match, expected %v, got %v",
				true, kernelCfg.Config.IgnoreIneligibleTable)
		}
		if !pie.Equals(tableCfg.FilterRule, kernelCfg.Config.Filter.Rules) {
			return fmt.Errorf("filter rules not match, expected %v, got %v",
				tableCfg.FilterRule, kernelCfg.Config.Filter.Rules)
		}
	}
	if kafkaCfg.TopicPartitionConfig != nil {
		topicCfg := kafkaCfg.TopicPartitionConfig
		if sinkURI.Query().Get("partition-num") != fmt.Sprintf("%d", topicCfg.PartitionNum) {
			return fmt.Errorf("partition-num not match, expected %d, got %s", topicCfg.PartitionNum,
				sinkURI.Query().Get("partition-num"))
		}
		if sinkURI.Query().Get("replication-factor") != fmt.Sprintf("%d", topicCfg.ReplicationFactor) {
			return fmt.Errorf("replication-factor not match, expected %d, got %s", topicCfg.PartitionNum,
				sinkURI.Query().Get("replication-factor"))
		}
		if sinkURI.Path != fmt.Sprintf("/%s", topicCfg.DefaultTopic) {
			return fmt.Errorf("default topic not match, expected %s, got %s", topicCfg.DefaultTopic, sinkURI.Path)
		}
		var topicValue string
		switch *topicCfg.DispatchType {
		case model.DataflowDispatchTypeDISPATCHTYPEONETOPIC:
			topicValue = ""
		case model.DataflowDispatchTypeDISPATCHTYPEBYTABLE:
			middle := "_"
			if topicCfg.Separator != "" {
				middle = topicCfg.Separator
			}
			topicValue = fmt.Sprintf("%s{schema}%s{table}%s", topicCfg.TopicPrefix, middle, topicCfg.TopicSuffix)
		case model.DataflowDispatchTypeDISPATCHTYPEBYDATABASE:
			topicValue = fmt.Sprintf("%s{schema}%s", topicCfg.TopicPrefix, topicCfg.TopicSuffix)
		}

		if topicCfg.PartitionDispatchers != nil {
			if len(topicCfg.PartitionDispatchers) != len(kernelCfg.Config.Sink.Dispatchers) {
				return fmt.Errorf("partition dispatchers length not match, expected %d, got %d",
					len(topicCfg.PartitionDispatchers), len(kernelCfg.Config.Sink.Dispatchers))
			}
			for i, apiDispatcher := range topicCfg.PartitionDispatchers {
				kernelDispatcher := kernelCfg.Config.Sink.Dispatchers[i]
				if !pie.Equals(apiDispatcher.Columns, kernelDispatcher.Columns) {
					return fmt.Errorf("partition dispatchers[%d] columns not match, expected %s, got %s",
						i, apiDispatcher.Columns, kernelDispatcher.Columns)
				}
				if topicValue != kernelDispatcher.Topic {
					return fmt.Errorf("partition dispatchers[%d] topic not match, expected %s, got %s",
						i, topicValue, kernelDispatcher.Topic)
				}
				if apiDispatcher.IndexName != kernelDispatcher.Index {
					return fmt.Errorf("partition dispatchers[%d] index name not match, expected %s, got %s",
						i, apiDispatcher.IndexName, kernelDispatcher.Index)
				}
				if !pie.Equals(apiDispatcher.Matcher, kernelDispatcher.Matcher) {
					return fmt.Errorf("partition dispatchers[%d] matcher not match, expected %s, got %s",
						i, apiDispatcher.Matcher, kernelDispatcher.Matcher)
				}
				if strings.ToLower(transferPartitionDispatcher(*apiDispatcher.PartitionType)) != strings.ToLower(kernelDispatcher.Partition) {
					return fmt.Errorf("partition dispatchers[%d] partition type not match, expected %s, got %s",
						i, *apiDispatcher.PartitionType, kernelDispatcher.Partition)
				}
			}
		}
	}

	return nil
}

const (
	KafkaVersion1xx = "1.0.0"
	KafkaVersion2xx = "2.0.0"
	KafkaVersion3xx = "3.0.0"

	PartitionTs         = "ts"
	PartitionTable      = "table"
	PartitionIndexValue = "index-value"
	PartitionColumns    = "columns"

	KafkaProtocolTypeCanalJson    = "canal-json"
	KafkaProtocolTypeOpenProtocol = "open-protocol"
	KafkaProtocolTypeAvro         = "avro"
	KafkaProtocolDebezium         = "debezium"
)

func transferKafkaVersion(kafkaVersion model.DataflowKafkaVersion) string {
	switch kafkaVersion {
	case model.DataflowKafkaVersionKAFKAVERSION0XX:
		return KafkaVersion1xx
	case model.DataflowKafkaVersionKAFKAVERSION2XX:
		return KafkaVersion2xx
	case model.DataflowKafkaVersionKAFKAVERSION3XX:
		return KafkaVersion3xx
	default:
		return ""
	}
}

func transferPartitionDispatcher(partition model.DataflowPartitionDispatcher) string {
	switch partition {
	case model.DataflowPartitionDispatcherTABLE:
		return PartitionTable
	case model.DataflowPartitionDispatcherINDEXVALUE:
		return PartitionIndexValue
	case model.DataflowPartitionDispatcherTS:
		return PartitionTs
	case model.DataflowPartitionDispatcherCOLUMNS:
		return PartitionColumns
	}
	return ""
}

func transferProtocol(protocol model.DataflowProtocol) string {
	switch protocol {
	case model.DataflowProtocolCANALJSON:
		return KafkaProtocolTypeCanalJson
	case model.DataflowProtocolOPENPROTOCOL:
		return KafkaProtocolTypeOpenProtocol
	case model.DataflowProtocolAVRO:
		return KafkaProtocolTypeAvro
	case model.DataflowProtocolDEBEZIUM:
		return KafkaProtocolDebezium
	}
	return ""
}