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.
ticdc(pulsar): write DDL events to pulsar (pingcap#9433)
ref pingcap#9413
- Loading branch information
1 parent
2af966e
commit a199d59
Showing
20 changed files
with
1,181 additions
and
46 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
113 changes: 113 additions & 0 deletions
113
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,113 @@ | ||
| // 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/log" | ||
| "github.com/pingcap/tiflow/cdc/model" | ||
| "github.com/pingcap/tiflow/pkg/config" | ||
| "github.com/pingcap/tiflow/pkg/sink/codec/common" | ||
| pulsarConfig "github.com/pingcap/tiflow/pkg/sink/pulsar" | ||
| ) | ||
|
|
||
| // 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 { | ||
| // call SyncSendMessage | ||
|
|
||
| log.Info("pulsarProducers SyncBroadcastMessage in") | ||
| 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 *pulsarConfig.Config, | ||
| 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 *pulsarConfig.Config, | ||
| 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] | ||
| } |
Oops, something went wrong.