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

Hey,

So here's what I think is happening. The keep alive timeout exactly the
same as the option in the slot.

So in

private boolean isTimeUpdate() {
  /* a value of 0 disables automatic updates */
  if ( updateInterval == 0 ) {
    return false;
  }
  long diff = System.currentTimeMillis() - lastStatusUpdate;
  return diff >= updateInterval;
}

change this to be


return diff >= updateInterval-100;

or something similar to get it to update before it times out on the server


Can you build the driver ? If not I'll send you a driver with some mods to
change

On Tue, 4 Sep 2018 at 08:40, Dave Cramer <[email protected]> wrote:

>
> 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
>>>>
>>>