postgresql Logical Replication Stream fails with “ Database connection failed when reading from copy”

Jammie <[email protected]>
Newsgroups gmane.comp.db.postgresql.jdbc
Message-ID <CAFt1pcpg-w_rJ56EmkPyfeZxU7uR-3QU93UyWmr=SnEfpQneqw@mail.gmail.com>
Hi I am reading from logical replication stream continuously in infinite
while loop. and I have another program that is continuously populating a
table in the same database. I notice that after some time (around 5-10
minutes). I always get the exception

org.postgresql.util.PSQLException: Database connection failed when
reading from copy
at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1035)
at org.postgresql.core.v3.CopyDualImpl.readFromCopy(CopyDualImpl.java:41)
at org.postgresql.core.v3.replication.V3PGReplicationStream.receiveNextData(V3PGReplicationStream.java:155)
at org.postgresql.core.v3.replication.V3PGReplicationStream.readInternal(V3PGReplicationStream.java:124)
at org.postgresql.core.v3.replication.V3PGReplicationStream.read(V3PGReplicationStream.java:70)
at com.datamirror.ts.scrapers.postgresqlscraper.PGStreamReceiver.main(PGStreamReceiver.java:60)
Caused by: java.net.SocketException: Connection reset
at java.net.SocketInputStream.read(SocketInputStream.java:220)
at java.net.SocketInputStream.read(SocketInputStream.java:152)
at org.postgresql.core.VisibleBufferedInputStream.readMore(VisibleBufferedInputStream.java:140)
at org.postgresql.core.VisibleBufferedInputStream.ensureBytes(VisibleBufferedInputStream.java:109)
at org.postgresql.core.VisibleBufferedInputStream.read(VisibleBufferedInputStream.java:67)
at org.postgresql.core.PGStream.receiveChar(PGStream.java:293)
at org.postgresql.core.v3.QueryExecutorImpl.processCopyResults(QueryExecutorImpl.java:1077)
at org.postgresql.core.v3.QueryExecutorImpl.readFromCopy(QueryExecutorImpl.java:1033)

Here is my sample program :

import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import org.postgresql.PGConnection;
import org.postgresql.PGProperty;
import org.postgresql.replication.LogSequenceNumber;
import org.postgresql.replication.PGReplicationStream;

public class PGStreamReceiver {
public static void main(String[] args) {
    try {
        LogSequenceNumber startLsn = LogSequenceNumber.valueOf("0/2D8D0F0");
         String url = "jdbc:postgresql://localhost:5432/postgres"
                 ;
         String user ="postgres";
          String password = "xxxx";
          Properties connectionProperties = new Properties();

          PGProperty.USER.set(connectionProperties, user);
          PGProperty.PASSWORD.set(connectionProperties, password);
          PGProperty.ASSUME_MIN_SERVER_VERSION.set(connectionProperties,
PostgreSQLConstants.MINIMUM_SUPPORTED_POSTGRESQL_VERSION);
          PGProperty.REPLICATION.set(connectionProperties,
PostgreSQLConstants.POSTGRESQL_REPLICATION_PROPERTY_VALUE);
          PGProperty.PREFER_QUERY_MODE.set(connectionProperties,
PostgreSQLConstants.POSTGRESQL_REPLICATION_PREFERRED_QUERY_MODE);
          Connection postgresSQLConnection =
DriverManager.getConnection(url, connectionProperties);
            PGConnection postgrePGConnectionWrapper =
postgresSQLConnection.unwrap(PGConnection.class);
            PGReplicationStream stream = postgrePGConnectionWrapper.
                     getReplicationAPI()
                     .replicationStream()
                     .logical()
                     .withSlotName("pvn")
                     .withStartPosition(startLsn)

.withSlotOption(PostgreSQLConstants.INCLUDE_XID_IN_STREAM_CHANGES,
true)

.withSlotOption(PostgreSQLConstants.INCLUDE_TIMESTAMP_IN_STREAM_CHANGES,
true)

.withSlotOption(PostgreSQLConstants.EXCLUDE_EMPTY_TRANSACTION_IN_CHANGES,
true)

//.withStatusInterval(PostgreSQLConstants.SERVER_FEEDBACK_TIME_INTERVAL_IN_SECONDS,
TimeUnit.SECONDS)
                     .withStatusInterval(60, TimeUnit.SECONDS)
                     .start();
            while(true)
            {
                ByteBuffer msg = stream.read();
                if(msg == null)
                {
                    return;
                }
                int offset = msg.arrayOffset();
                byte[] source = msg.array();
                int length = source.length - offset;
                String logData = new String(source, offset, length);
                System.out.print("1");
            }
        } catch (SQLException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}

Any help is appreciated !! I am using postgres JDBC driver 42.2.2.
Reference:
https://stackoverflow.com/questions/52147844/postgresql-logical-replication-stream-fails-with-database-connection-failed-whe

Regards
Shailesh
lmpx.com only provides a reader for public news (NNTP) servers. It is not affiliated with the servers or forums shown here and is not responsible for the content of articles, which is written by their respective authors.