0

Describe the issue We are facing issue of pg replication slot not listen database wal events. Once pg database restarted and we attach

  • Driver Version: 42.2.6
  • Java Version: JDK1.8
  • OS Version: Window 10
  • PostgreSQL Version: 9.6

Steps to reproduce the behaviour:

Code using the sample from the JDBC driver's documentation

Created replication connection logical with test decode

String url = "jdbc:postgresql://dbhist8:port/" + db;
Properties props = new Properties();
PGProperty.USER.set(props, "postgres");
PGProperty.PASSWORD.set(props, "postgres");
PGProperty.ASSUME_MIN_SERVER_VERSION.set(props, "10.5");
PGProperty.REPLICATION.set(props, "database");
PGProperty.PREFER_QUERY_MODE.set(props, "simple");
Connection slotVerifyConnection = null;
Connection slotConnection = null;
try {
//TODO connection pool
slotVerifyConnection = connect(url, "postgres", "postgres");
slotConnection = DriverManager.getConnection(url, props);
Statement statement = slotVerifyConnection.createStatement();
ResultSet result = statement.executeQuery("select restart_lsn,active,active_pid from pg_replication_slots where slot_type = 'logical' and slot_name ='" + slotName + "'");
while (result.next()) {

            lastLSNNumber = LogSequenceNumber.valueOf(result.getString(1));

        }

        Connection con = DriverManager.getConnection(url, props);
        PGConnection replConnection = con.unwrap(PGConnection.class);

        PGReplicationStream stream = null;


        if (lastLSNNumber != null) {
            stream =
                    replConnection.getReplicationAPI()
                            .replicationStream()
                            .logical()
                            .withSlotName(slotName)
                            //.withSlotOption("include-xids", true)
                             //.withSlotOption("skip-empty-xacts", true)
                            .withStartPosition(lastLSNNumber)

                            // .withStatusInterval(20, TimeUnit.SECONDS)
                            .withStatusInterval(Integer.parseInt("0"), TimeUnit.SECONDS).start();
        } else {

            **ReplicationSlotInfo info=       replConnection.getReplicationAPI()
                    .createReplicationSlot()
                    .logical()
                    .withSlotName(slotName)
                    .withOutputPlugin(wal2json)
                    .make();**
            info.getConsistentPoint();
            stream =
                    replConnection.getReplicationAPI()
                            .replicationStream()
                            .logical()
                            .withSlotName(slotName)
                        //    .withSlotOption("include-xids", false)
                       //     .withSlotOption("skip-empty-xacts", true)
                            .withStatusInterval(0, TimeUnit.SECONDS)
                            .start();
        }
        while (true) {
            //non blocking receive message
            ByteBuffer msg = stream.read();

            if (msg == null) {
                TimeUnit.MILLISECONDS.sleep(10L);
                continue;
            }

            int offset = msg.arrayOffset();
            byte[] source = msg.array();
            int length = source.length - offset;
            System.out.println(new String(source, offset, length));
            String sql=new String(source, offset, length);
            //QueryUtils.getQueries(sql);
            if(stream.getLastReceiveLSN()!=null) {
                stream.setAppliedLSN(stream.getLastReceiveLSN());
                stream.setFlushedLSN(stream.getLastReceiveLSN());
                stream.forceUpdateStatus();
            }else{
                System.out.println("NULL value");
            }
        }

    } catch (Exception ex) {
ex.printStackTrace();
}
}
  • And what exactly is your question? – a_horse_with_no_name Aug 7 at 19:02
  • Your link is to the section on physical replication, but you seem to be talking about logical replication. Also, please post code that compiles. – jjanes Aug 8 at 1:02
  • guys, issue is clear using maven dependency and example jdbc.postgresql.org/documentation/head/… we are trying run logical replication but slots created on db, BUT ISSUE WE ARE NOT GETTING events of wals – vaibhav.patil Aug 8 at 6:15
  • I am able to fix issue , this is due to jdbc wal slot creation issue, I am creating slot now using SQL command not with JDBC api which fixed my issue – vaibhav.patil Aug 15 at 3:36
0

I have fixed issue using SQL slot creation sql command. Removed below code which blocks executions ReplicationSlotInfo info= replConnection.getReplicationAPI() .createReplicationSlot() .logical() .withSlotName(slotName) .withOutputPlugin(wal2json) .make();

And my issue was fixed. Let me know if anyone face same issue I can help.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.