support DisableReplication

apache · Jun 16, 2021 · bc70f04 · bc70f04
1 parent f17deac
commit bc70f04
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 0 deletions.
diff --git a/pulsar/message.go b/pulsar/message.go
@@ -49,6 +49,9 @@ type ProducerMessage struct {
 	// ReplicationClusters override the replication clusters for this message.
 	ReplicationClusters []string
 
+	// Disable the replication for this message
+	DisableReplication bool
+
 	// SequenceID set the sequence id to assign to the current message
 	SequenceID *int64
 

diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
@@ -392,6 +392,11 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	if !sendAsBatch {
 		p.internalFlushCurrentBatch()
 	}
+
+	if msg.DisableReplication {
+		msg.ReplicationClusters = [] string{"__local__"}
+	}
+
 	added := p.batchBuilder.Add(smm, p.sequenceIDGenerator, payload, request,
 		msg.ReplicationClusters, deliverAt)
 	if !added {