mq

Documentation

Index

• type DDLDispatchRule
• type DDLSink
  • func NewKafkaDDLSink(ctx context.Context, changefeedID model.ChangeFeedID, sinkURI *url.URL, ...) (_ *DDLSink, err error)
  • func NewPulsarDDLSink(ctx context.Context, changefeedID model.ChangeFeedID, sinkURI *url.URL, ...) (_ *DDLSink, err error)
  • func (k *DDLSink) Close()
  • func (k *DDLSink) WriteCheckpointTs(ctx context.Context, ts uint64, tables []*model.TableInfo) error
  • func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error

Types

type DDLDispatchRule

type DDLDispatchRule int

DDLDispatchRule is the dispatch rule for DDL event.

const (
	// PartitionZero means the DDL event will be dispatched to partition 0.
	// NOTICE: Only for canal and canal-json protocol.
	PartitionZero DDLDispatchRule = iota
	// PartitionAll means the DDL event will be broadcast to all the partitions.
	PartitionAll
)

type DDLSink

type DDLSink struct {
	// contains filtered or unexported fields
}

DDLSink is a sink that sends DDL events to the MQ system.

func NewKafkaDDLSink

func NewKafkaDDLSink(
	ctx context.Context,
	changefeedID model.ChangeFeedID,
	sinkURI *url.URL,
	replicaConfig *config.ReplicaConfig,
	factoryCreator kafka.FactoryCreator,
	producerCreator ddlproducer.Factory,
) (_ *DDLSink, err error)

NewKafkaDDLSink will verify the config and create a Kafka DDL Sink.

func NewPulsarDDLSink

func NewPulsarDDLSink(
	ctx context.Context,
	changefeedID model.ChangeFeedID,
	sinkURI *url.URL,
	replicaConfig *config.ReplicaConfig,
	pulsarTopicManagerCreator manager.PulsarTopicManager,
	clientCreator pulsarConfig.FactoryCreator,
	producerCreator ddlproducer.PulsarFactory,
) (_ *DDLSink, err error)

NewPulsarDDLSink will verify the config and create a Pulsar DDL Sink.

func (*DDLSink) Close

func (k *DDLSink) Close()

Close closes the sink.

func (*DDLSink) WriteCheckpointTs

func (k *DDLSink) WriteCheckpointTs(ctx context.Context,
	ts uint64, tables []*model.TableInfo,
) error

WriteCheckpointTs sends the checkpoint ts to the MQ system.

func (*DDLSink) WriteDDLEvent

func (k *DDLSink) WriteDDLEvent(ctx context.Context, ddl *model.DDLEvent) error

WriteDDLEvent encodes the DDL event and sends it to the MQ system.