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 <davec(at)postgresintl(dot)com> wrote:

>
> On Mon, 3 Sep 2018 at 21:03, Jammie <shailesh(dot)jamloki(at)gmail(dot)com> 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 <pg(at)fastcrypt(dot)com> 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
>>>
>>> davec(at)postgresintl(dot)com
>>> www.postgresintl.com
>>>
>>>
>>> On Mon, 3 Sep 2018 at 06:34, Jammie <shailesh(dot)jamloki(at)gmail(dot)com> 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
>>>>
>>>