forked from pingcap/tiflow
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
This is an automated cherry-pick of pingcap#9691
Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>
- Loading branch information
1 parent
5fe84bb
commit 4f8aed2
Showing
17 changed files
with
1,148 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
108 changes: 108 additions & 0 deletions
108
cdc/sink/ddlsink/mq/ddlproducer/pulsar_ddl_mock_producer.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| // Copyright 2023 PingCAP, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package ddlproducer | ||
|
|
||
| import ( | ||
| "context" | ||
|
|
||
| "github.com/apache/pulsar-client-go/pulsar" | ||
| "github.com/pingcap/tiflow/cdc/model" | ||
| "github.com/pingcap/tiflow/pkg/config" | ||
| "github.com/pingcap/tiflow/pkg/sink/codec/common" | ||
| ) | ||
|
|
||
| // Assert DDLEventSink implementation | ||
| var _ DDLProducer = (*PulsarMockProducers)(nil) | ||
|
|
||
| // PulsarMockProducers is a mock pulsar producer | ||
| type PulsarMockProducers struct { | ||
| events map[string][]*pulsar.ProducerMessage | ||
| } | ||
|
|
||
| // SyncBroadcastMessage pulsar consume all partitions | ||
| func (p *PulsarMockProducers) SyncBroadcastMessage(ctx context.Context, topic string, | ||
| totalPartitionsNum int32, message *common.Message, | ||
| ) error { | ||
| return p.SyncSendMessage(ctx, topic, totalPartitionsNum, message) | ||
| } | ||
|
|
||
| // SyncSendMessage sends a message | ||
| // partitionNum is not used,pulsar consume all partitions | ||
| func (p *PulsarMockProducers) SyncSendMessage(ctx context.Context, topic string, | ||
| partitionNum int32, message *common.Message, | ||
| ) error { | ||
| data := &pulsar.ProducerMessage{ | ||
| Payload: message.Value, | ||
| Key: message.GetPartitionKey(), | ||
| } | ||
| p.events[topic] = append(p.events[topic], data) | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| // NewMockPulsarProducer creates a pulsar producer | ||
| func NewMockPulsarProducer( | ||
| ctx context.Context, | ||
| changefeedID model.ChangeFeedID, | ||
| pConfig *config.PulsarConfig, | ||
| client pulsar.Client, | ||
| ) (*PulsarMockProducers, error) { | ||
| return &PulsarMockProducers{ | ||
| events: map[string][]*pulsar.ProducerMessage{}, | ||
| }, nil | ||
| } | ||
|
|
||
| // NewMockPulsarProducerDDL creates a pulsar producer for DDLProducer | ||
| func NewMockPulsarProducerDDL( | ||
| ctx context.Context, | ||
| changefeedID model.ChangeFeedID, | ||
| pConfig *config.PulsarConfig, | ||
| client pulsar.Client, | ||
| sinkConfig *config.SinkConfig, | ||
| ) (DDLProducer, error) { | ||
| return NewMockPulsarProducer(ctx, changefeedID, pConfig, client) | ||
| } | ||
|
|
||
| // GetProducerByTopic returns a producer by topic name | ||
| func (p *PulsarMockProducers) GetProducerByTopic(topicName string) (producer pulsar.Producer, err error) { | ||
| return producer, nil | ||
| } | ||
|
|
||
| // Close close all producers | ||
| func (p *PulsarMockProducers) Close() { | ||
| p.events = make(map[string][]*pulsar.ProducerMessage) | ||
| } | ||
|
|
||
| // Flush waits for all the messages in the async producer to be sent to Pulsar. | ||
| // Notice: this method is not thread-safe. | ||
| // Do not try to call AsyncSendMessage and Flush functions in different threads, | ||
| // otherwise Flush will not work as expected. It may never finish or flush the wrong message. | ||
| // Because inflight will be modified by mistake. | ||
| func (p *PulsarMockProducers) Flush(ctx context.Context) error { | ||
| return nil | ||
| } | ||
|
|
||
| // GetAllEvents returns the events received by the mock producer. | ||
| func (p *PulsarMockProducers) GetAllEvents() []*pulsar.ProducerMessage { | ||
| var events []*pulsar.ProducerMessage | ||
| for _, v := range p.events { | ||
| events = append(events, v...) | ||
| } | ||
| return events | ||
| } | ||
|
|
||
| // GetEvents returns the event filtered by the key. | ||
| func (p *PulsarMockProducers) GetEvents(topic string) []*pulsar.ProducerMessage { | ||
| return p.events[topic] | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,106 @@ | ||
| // Copyright 2023 PingCAP, Inc. | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| package mq | ||
|
|
||
| import ( | ||
| "context" | ||
| "net/url" | ||
| "time" | ||
|
|
||
| "github.com/pingcap/errors" | ||
| "github.com/pingcap/log" | ||
| "github.com/pingcap/tiflow/cdc/model" | ||
| "github.com/pingcap/tiflow/cdc/sink/ddlsink/mq/ddlproducer" | ||
| "github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher" | ||
| "github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/manager" | ||
| "github.com/pingcap/tiflow/cdc/sink/util" | ||
| "github.com/pingcap/tiflow/pkg/config" | ||
| cerror "github.com/pingcap/tiflow/pkg/errors" | ||
| "github.com/pingcap/tiflow/pkg/sink/codec/builder" | ||
| pulsarConfig "github.com/pingcap/tiflow/pkg/sink/pulsar" | ||
| tiflowutil "github.com/pingcap/tiflow/pkg/util" | ||
| "go.uber.org/zap" | ||
| ) | ||
|
|
||
| // NewPulsarDDLSink will verify the config and create a Pulsar DDL Sink. | ||
| 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) { | ||
| log.Info("Starting pulsar DDL producer ...", | ||
| zap.String("namespace", changefeedID.Namespace), | ||
| zap.String("changefeed", changefeedID.ID)) | ||
|
|
||
| defaultTopic, err := util.GetTopic(sinkURI) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| protocol, err := util.GetProtocol(tiflowutil.GetOrZero(replicaConfig.Sink.Protocol)) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| pConfig, err := pulsarConfig.NewPulsarConfig(sinkURI, replicaConfig.Sink.PulsarConfig) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| log.Info("Try to create a DDL sink producer", | ||
| zap.String("changefeed", changefeedID.String())) | ||
|
|
||
| // NewEventRouter | ||
| eventRouter, err := dispatcher.NewEventRouter(replicaConfig, defaultTopic, sinkURI.Scheme) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| encoderConfig, err := util.GetEncoderConfig(changefeedID, | ||
| sinkURI, protocol, replicaConfig, config.DefaultMaxMessageBytes) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| encoderBuilder, err := builder.NewRowEventEncoderBuilder(ctx, encoderConfig) | ||
| if err != nil { | ||
| return nil, cerror.WrapError(cerror.ErrKafkaInvalidConfig, err) | ||
| } | ||
|
|
||
| start := time.Now() | ||
| client, err := clientCreator(pConfig, changefeedID, replicaConfig.Sink) | ||
| if err != nil { | ||
| log.Error("DDL sink producer client create fail", zap.Error(err)) | ||
| return nil, cerror.WrapError(cerror.ErrPulsarNewClient, err) | ||
| } | ||
|
|
||
| p, err := producerCreator(ctx, changefeedID, pConfig, client, replicaConfig.Sink) | ||
| log.Info("DDL sink producer client created", zap.Duration("duration", time.Since(start))) | ||
| if err != nil { | ||
| return nil, cerror.WrapError(cerror.ErrPulsarNewProducer, err) | ||
| } | ||
|
|
||
| topicManager, err := pulsarTopicManagerCreator(pConfig, client) | ||
| if err != nil { | ||
| return nil, errors.Trace(err) | ||
| } | ||
|
|
||
| s := newDDLSink(ctx, changefeedID, p, nil, topicManager, eventRouter, encoderBuilder, protocol) | ||
|
|
||
| return s, nil | ||
| } |
Oops, something went wrong.