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

On Mon, 3 Sep 2018 at 21:03, Jammie <[email protected]> wrote:

> Hi Dave,
> thanks for the response !!
>
> 1) I have tried latest driver 42.2.5 as well the same exception can be
> reproduced easily.
> 2) I have tried postgres V10 version along with 9.6.4 same issue could be
> reproduced.
> The server side log says :
> *2018-09-02 23:40:52 EDT LOG:  could not receive data from client: An
> existing connection was forcibly closed by the remote host.*
>
> In one of the instance I had enabled JDBC trace and this is what the last
> few lines are :
>
> *Sep 03, 2018 3:37:52 AM org.postgresql.core.v3.*
>
> *QueryExecutorImpl receiveRFQFINEST:  <=BE ReadyForQuery(I)Sep 03, 2018
> 3:37:52 AM org.postgresql.jdbc.*
>
> *PgResultSet getIntFINEST:   getInt columnIndex: 1Sep 03, 2018 3:37:52 AM
> org.postgresql.jdbc.*
>
> *PgConnection setAutoCommitFINE:   setAutoCommit = falseSep 03, 2018
> 3:37:53 AM org.postgresql.core.**QueryExecutorBase close*
>
> I also see Socket reset error some times :
> ---------------------
> *Sep 03, 2018 3:29:42 AM org.postgresql.core.v3.*
>
> *QueryExecutorImpl processCopyResultsFINEST:  <=BE CopyDataSep 03, 2018
> 3:29:42 AM org.postgresql.core.v3.**replication.*
>
> *V3PGReplicationStream processKeepAliveMessageFINEST:   <=BE
> Keepalive(lastServerWal: 0/2DD0218, clock: 9/3/18 3:29 AM needReply:
> false)Sep 03, 2018 3:30:40 AM org.postgresql.core.*
>
> *QueryExecutorBase closeFINEST:  FE=> TerminateSep 03, 2018 3:30:40 AM
> org.postgresql.core.*
>
>
> *QueryExecutorBase closeFINEST: Discarding IOException on
> close:java.net.SocketException: Connection reset by peer: socket write
> error    at java.net.SocketOutputStream.*
> *socketWrite0(Native Method)    at java.net.SocketOutputStream.*
> *socketWrite(*
> *SocketOutputStream.java:122)    at java.net.SocketOutputStream.*
> *write(SocketOutputStream.java:*
> *166)    at java.io.BufferedOutputStream.**flushBuffer(*
> *BufferedOutputStream.java:93)    at java.io.BufferedOutputStream.*
> *flush(BufferedOutputStream.*
> *java:151)    at org.postgresql.core.PGStream.*
> *flush(PGStream.java:514)    at org.postgresql.core.*
> *QueryExecutorBase.close(*
> *QueryExecutorBase.java:136)    at org.postgresql.jdbc.*
> *PgConnection.close(*
> *PgConnection.java:659)    at com.datamirror.ts.scrapers.*
> *postgresqlscraper.**PostgreSQLLogicalLogReaderImpl**.readLog(*
> *PostgreSQLLogicalLogReaderImpl*
> *.java:208)    at com.datamirror.ts.scrapers.**postgresqlscraper.*
> *PostgreSQLLogReader.readLog(*
> *PostgreSQLLogReader.java:174)    at com.datamirror.ts.scrapers.*
> *cdc.LogReader.execute(*
> *LogReader.java:341)    at com.datamirror.ts.scrapers.**cdc.LogReader$*
> *LogReaderPipelineJob.execute(*
> *LogReader.java:114)    at com.datamirror.ts.engine.*
> *component.PipelineThread.**runThread(PipelineThread.java:*
> *217)    at com.datamirror.ts.util.**TsThread.run(TsThread.java:*
>
>
> *130)Sep 03, 2018 3:30:40 AM org.postgresql.Driver connectFINEST:  FE=>
> Terminate*
> -------------------------
> 3) Other than this I have tried to set PGProperty TCP_KEEP_ALIVE to true
> and SOCKET_TIMEOUT as 0.
> Also SetStatusInterval of the stream also to 0 but these did not help the
> purpose.
>
> The issue is very consistent and can be reproduced easily.
>
> Regards
> Shailesh
>

Pretty sure I know how to fix this,

Give me a few hours..

>
>
>
> On Tue, Sep 4, 2018 at 12:51 AM Dave Cramer <[email protected]> wrote:
>
>> Hi,
>>
>> Off the top of my head I don't have any advice. Can you try the latest
>> driver though and see if it still exists.
>>
>> Are there any messages on the server saying that it closed the connection
>> ?
>>
>> Dave Cramer
>>
>> [email protected]
>> www.postgresintl.com
>>
>>
>> On Mon, 3 Sep 2018 at 06:34, Jammie <[email protected]> wrote:
>>
>>> 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
>>>
>>