public inbox for [email protected]  

help / color / mirror / Atom feed

From: Christophe Pettus <[email protected]>
To: [email protected]
Subject: Multiple message delivery on logical replication
Date: Fri, 2 Nov 2018 18:28:51 -0700
Message-ID: <[email protected]> (raw)

I'm working with the logical replication support in psycopg2, and have found something surprising... this may be my error, of course!

My sample program is below.  It works wonderfully, but in the case when it starts, it re-receives the last message that it handled, even with flushing it.

Example:

postgres@localhost:~/wal2pubsub$ python waltest.py 
{"change":[{"kind":"insert","schema":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"columnvalues":[6]},{"kind":"insert","schema":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"columnvalues":[7]}]}
^C
postgres@localhost:~/wal2pubsub$ python waltest.py 
{"change":[{"kind":"insert","schema":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"columnvalues":[6]},{"kind":"insert","schema":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"columnvalues":[7]}]}

There was no database activity in that period; it just replayed the same message.  Shouldn't it have flushed to the end of the WAL stream and not reprocessed the last message?

--

import psycopg2
from psycopg2.extras import LogicalReplicationConnection, REPLICATION_LOGICAL

conn = psycopg2.connect('dbname=postgres', connection_factory=LogicalReplicationConnection)
cur = conn.cursor()

cur.start_replication(slot_name='test_slot', slot_type=REPLICATION_LOGICAL)

from select import select
from datetime import datetime

def consume(msg):
    print(msg.payload)
    msg.cursor.send_feedback(flush_lsn=msg.data_start)

try:
    cur.consume_stream(consume)
except:
    pass

--
-- Christophe Pettus
   [email protected]

view thread (3+ messages)  latest in thread

reply

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Reply to all the recipients using the --to and --cc options:
  reply via email

  To: [email protected]
  Cc: [email protected]
  Subject: Re: Multiple message delivery on logical replication
  In-Reply-To: <[email protected]>

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox