From: Christophe Pettus <xof@thebuild.com>
Content-Type: text/plain;
	charset=us-ascii
Content-Transfer-Encoding: quoted-printable
Mime-Version: 1.0 (Mac OS X Mail 11.5 \(3445.9.1\))
Subject: Multiple message delivery on logical replication
Message-Id: <A7E70A6B-04AB-4D26-BD7E-5E44DC0DA96D@thebuild.com>
Date: Fri, 2 Nov 2018 18:28:51 -0700
To: psycopg@postgresql.org
Precedence: bulk

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=20
=
{"change":[{"kind":"insert","schema":"public","table":"x","columnnames":["=
i"],"columntypes":["integer"],"columnvalues":[6]},{"kind":"insert","schema=
":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"colu=
mnvalues":[7]}]}
^C
postgres@localhost:~/wal2pubsub$ python waltest.py=20
=
{"change":[{"kind":"insert","schema":"public","table":"x","columnnames":["=
i"],"columntypes":["integer"],"columnvalues":[6]},{"kind":"insert","schema=
":"public","table":"x","columnnames":["i"],"columntypes":["integer"],"colu=
mnvalues":[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 =3D psycopg2.connect('dbname=3Dpostgres', =
connection_factory=3DLogicalReplicationConnection)
cur =3D conn.cursor()

cur.start_replication(slot_name=3D'test_slot', =
slot_type=3DREPLICATION_LOGICAL)

from select import select
from datetime import datetime

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

try:
    cur.consume_stream(consume)
except:
    pass

--
-- Christophe Pettus
   xof@thebuild.com