Merge pull request #61 from replicase/feature/add-server-time-to-pg-s…

…ource-checkpoint feat: add server time info to the pg source checkpoints
replicase · Jul 24, 2024 · ec7ea6d · ec7ea6d
2 parents 34cec4b + e4b118d
commit ec7ea6d
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 6 deletions.
diff --git a/pkg/cursor/main.go b/pkg/cursor/main.go
@@ -4,15 +4,17 @@ import (
  "errors"
  "strconv"
  "strings"
+ "time"
 
  "github.com/apache/pulsar-client-go/pulsar"
  "github.com/jackc/pglogrepl"
 )
 
 type Checkpoint struct {
- LSN uint64
- Seq uint32
- Data []byte
+ LSN uint64
+ Seq uint32
+ Data []byte
+ ServerTime time.Time
 }
 
 func (cp *Checkpoint) Equal(cp2 Checkpoint) bool {

diff --git a/pkg/source/postgres.go b/pkg/source/postgres.go
@@ -189,8 +189,15 @@ func (p *PGXSource) fetching(ctx context.Context) (change Change, err error) {
  p.nextReportTime = time.Time{}
  }
  change = Change{
- Checkpoint: cursor.Checkpoint{LSN: uint64(pkm.ServerWALEnd)},
- Message: &pb.Message{Type: &pb.Message_KeepAlive{KeepAlive: &pb.KeepAlive{}}},
+ Checkpoint: cursor.Checkpoint{
+ LSN: uint64(pkm.ServerWALEnd),
+ ServerTime: pkm.ServerTime,
+ },
+ Message: &pb.Message{
+ Type: &pb.Message_KeepAlive{
+ KeepAlive: &pb.KeepAlive{},
+ },
+ },
  }
  case pglogrepl.XLogDataByteID:
  xld, err := pglogrepl.ParseXLogData(msg.Data[1:])
@@ -221,7 +228,7 @@ func (p *PGXSource) fetching(ctx context.Context) (change Change, err error) {
  p.currentSeq++
  }
  change = Change{
- Checkpoint: cursor.Checkpoint{LSN: p.currentLsn, Seq: p.currentSeq},
+ Checkpoint: cursor.Checkpoint{LSN: p.currentLsn, Seq: p.currentSeq, ServerTime: xld.ServerTime},
  Message: m,
  }
  if !p.first {