Hi all,
I am developing an application which connects to a logical replication slot, to
consume the WAL events. These WAL events are then forwarded to a MQ broker.
The heart of the code can be seen as:
while (true) {
Connection connection = null;
PGReplicationStream stream = null;
try {
connection =
DriverManager.getConnection("jdbc:postgresql://localhost:5432/db", properties);
stream =
connection.unwrap(PGConnection.class).getReplicationAPI().replicationStream().logical().withSlotName("slot").start();
while (true) {
final ByteBuffer buffer = stream.read();
// ... MQ logic here ... omitted ...
stream.setAppliedLSN(stream.getLastReceiveLSN());
stream.setFlushedLSN(stream.getLastReceiveLSN());
}
} catch (final SQLException e) {
// ... log exception ... omitted ...
} finally {
// ... close stream and connection ... omitted ...
}
}
I notice some behavior which I cannot explain and would like to understand so I
can alter my code:
- When I restart the application, I notice that the application is retrieving
the last event from the previous run again. The result is that this event is
sent twice to the MQ broker after a restart of the application. Why is that?
Isn't calling `setAppliedLSN(stream.getLastReceiveLSN())` and/or
`setFlushedLSN(stream.getLastReceiveLSN())` enough to acknowledge an event, so
it will removed from the WAL log and it will not be resent?
- When receiving an event, the corresponding LSN from that event (which is sent
in the payload) is not the same as the result of `stream.getLastReceivedLSN()`.
Why is that? Which one should I use? Maybe this is correlated to my first
question.
- What is the difference between `setAppliedLSN(LSN)` and `setFlushedLSN(LSN)`?
The Javadocs are not really helpful here.
FYI, I also asked this question on
https://stackoverflow.com/questions/46301578/postgres-jdbc-logical-replication-lsn-feedback.
Thanks a lot and kind regards,
Yason TR
--
Sent via pgsql-general mailing list (pgsql-general@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-general
