fix: according to feedback

replicase · Jan 21, 2024 · 193fedc · 193fedc
1 parent a8199d2
commit 193fedc
Showing 1 changed file with 3 additions and 0 deletions.
diff --git a/pkg/sink/pulsar.go b/pkg/sink/pulsar.go
@@ -127,6 +127,9 @@ func (p *PulsarSink) Apply(changes chan source.Change) chan cursor.Checkpoint {
 		}
 
 		if change.Message.GetKeepAlive() != nil {
+			if err := p.producer.Flush(); err != nil {
+				return err
+			}
 			committed <- change.Checkpoint
 			return nil
 		}