postgresql Поток логической репликации завершается с ошибкой «Ошибка подключения к базе данных при чтении из копии» - PullRequest
Новая
       29

postgresql Поток логической репликации завершается с ошибкой «Ошибка подключения к базе данных при чтении из копии»

0 голосов
/ 03 сентября 2018

Привет, я непрерывно читаю поток логической репликации в бесконечном цикле и у меня есть другая программа, которая постоянно заполняет таблицу в той же базе данных. Я заметил это через некоторое время (около 5-10 минут). Я всегда получаю исключение 

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)

Вот мой пример программы: 

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();
        }
    }

}

Любая помощь приветствуется! Я использую драйвер JDBC Postgres 42.2.2.

...