public inbox for [email protected]
help / color / mirror / Atom feedMultiple message delivery on logical replication
3+ messages / 2 participants
[nested] [flat]
* Multiple message delivery on logical replication
@ 2018-11-03 01:28 Christophe Pettus <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: Christophe Pettus @ 2018-11-03 01:28 UTC (permalink / raw)
To: psycopg
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]
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Multiple message delivery on logical replication
@ 2018-11-03 01:31 Jonathan S. Katz <[email protected]>
parent: Christophe Pettus <[email protected]>
0 siblings, 1 reply; 3+ messages in thread
From: Jonathan S. Katz @ 2018-11-03 01:31 UTC (permalink / raw)
To: Christophe Pettus <[email protected]>; +Cc: psycopg
> On Nov 2, 2018, at 9:28 PM, Christophe Pettus <[email protected]> wrote:
>
> 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.
I’ve had this happen as well. IIRC it was when I was using it in async
mode with wal2json. From the investigating I did at the time, I thought
it was a bug with wal2json instead of psycopg2, but it was awhile ago
and my memory is fuzzy :(
Jonathan
Attachments:
[application/pgp-signature] signature.asc (833B, ../../[email protected]/2-signature.asc)
download
^ permalink raw reply [nested|flat] 3+ messages in thread
* Re: Multiple message delivery on logical replication
@ 2018-11-03 01:36 Christophe Pettus <[email protected]>
parent: Jonathan S. Katz <[email protected]>
0 siblings, 0 replies; 3+ messages in thread
From: Christophe Pettus @ 2018-11-03 01:36 UTC (permalink / raw)
To: psycopg
> On Nov 2, 2018, at 18:31, Jonathan S. Katz <[email protected]> wrote:
> I thought it was a bug with wal2json instead of psycopg2 [...]
That appears to be the case; test_decoding doesn't show the same behavior.
--
-- Christophe Pettus
[email protected]
^ permalink raw reply [nested|flat] 3+ messages in thread
end of thread, other threads:[~2018-11-03 01:36 UTC | newest]
Thread overview: 3+ messages (download: mbox mbox.gz follow: Atom feed)
-- links below jump to the message on this page --
2018-11-03 01:28 Multiple message delivery on logical replication Christophe Pettus <[email protected]>
2018-11-03 01:31 ` Jonathan S. Katz <[email protected]>
2018-11-03 01:36 ` Christophe Pettus <[email protected]>
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox