help me fix this program : ReqlDriverError: Response pump closed

[doantrungvn]

I am learning rethinkdb and I get the following error. After I started my application, I could add the record to table success, but after 15/20 minutes later, if I add the record, the error :
c.r.g.e.ReqlDriverError: Response pump closed. at c.r.n.DefaultConnectionFactory$ThreadResponsePump.await(DefaultConnectionFactory.java:214) at c.r.net.Connection.sendQuery(Connection.java:349) at c.r.net.Connection.runQuery(Connection.java:384) at c.r.net.Connection.runAsync(Connection.java:166) at c.r.net.Connection.run(Connection.java:185).

My code here:

@Override
public void run () {
    while (true) {
        try {
            logger.info ("START RETHINKDB & CHANGEFEED");
            conn = r.connection ()
                    .hostname (configParam.hostRethink)
                    .port (configParam.portRethink)
                    .db (configParam.dbRethink)
                    .user (configParam.userRethink, configParam.passRethink)
                    .connect ();

            // Changefeeds: subscribe to a feed by calling changes on a table
            Result <Object> result = r.table ("order")
                    .filter (
                            new ReqlFunction1 () {
                                @Override
                                public Object apply (ReqlExpr row) {
                                    return row.g ("status"). eq ("APPROVED");
                                }
                            }). changes (). optArg ("include_types", true) .run (conn);
            for (Object change: result) {
                logger.info (change.toString ());
                superMarketService.updateApprovedSuperMarketOrder (change);
            }
            logger.info ("END RETHINKDB & CHANGEFEED SUCCESS");
        } catch (Exception e) {
            e.printStackTrace ();
        }
    }
}

public void addOrder (JSONSuperMarket json) {
    if (conn == null) {
        logger.error ("=============== RETHINK DIE =========");
    }

    Result <Object> result = r.table ("order"). Insert (r.array (
            r.hashMap ("order_code", json.getOrderCode ())
                    .with ("cash_id", json.getCashId ())
                    .with ("merchant_id", json.getMerchantId ())
                    .with ("amount", json.getAmount ())
                    .with ("status", "PENDING")
                    .with ("description", json.getDescription ())
                    .with ("created_date", r.now (). inTimezone ("+ 07:00"). toIso8601 ())
    )). run (conn);
    logger.info (result.toString ());
}

Is there any error in my code, pls help me :(
I use Java 8 + RethinkDb 2.4.2.

[gabor-boros]

What is the RethinkDB version you have? I guess it is 2.4.0, right?

@adriantodt

[doantrungvn]

What is the RethinkDB version you have? I guess it is 2.4.0, right?

i use rethinkdb 2.4.2

[NotJustAnna]

I need the complete log, to be honest. I have my thoughts on how this happened, but just a stacktrace isn't enough. Post the log into a service such as Pastebin or GitHub's Gist.

Additionally, I'm doing a hotfix branch that should add the shutdown reason.

I can give you a bit of code review, though. This code is not "good" and I'm not expecting it to have any sort of consistent behaviour. You're creating a new connection every time an exception occurs,

[doantrungvn]

I need the complete log, to be honest. I have my thoughts on how this happened, but just a stacktrace isn't enough. Post the log into a service such as Pastebin or GitHub's Gist.

Additionally, I'm doing a hotfix branch that should add the shutdown reason.

I can give you a bit of code review, though. This code is not "good" and I'm not expecting it to have any sort of consistent behaviour. You're creating a new connection every time an exception occurs,

@adriantodt ,
remind my problem : After I started my application, I could add the record to table success, but after 15/20 minutes later, if I add the record -> the error Response pump closed

i send my code and the error, pls help me.

my code : https://gist.github.com/doantrungvn/10d813628a0a8490a6318e7197a8ec64
error : https://gist.github.com/doantrungvn/94d16b02a2984ff10b62eea40d82bdf6

[doantrungvn]

@adriantodt i waiting for u

[martinpaljak]

Also had this issue. I think I "fixed" this by replacing a thread-local connection with a getConnection() method that checks if connection isOpen() and does a connect() if not. This far has not happened again.

[martinpaljak]

Must take back my previous comment.

15:00:45:192 [whois-server] [INFO] DB - Reconnecting to DB
15:00:45:192 [whois-server] [INFO] DB - Closing failed, ignoring
com.rethinkdb.gen.exc.ReqlDriverError: Client not connected.
	at com.rethinkdb.net.Connection.sendQuery(Connection.java:449)
	at com.rethinkdb.net.Connection.runQuery(Connection.java:504)
	at com.rethinkdb.net.Connection.noreplyWaitAsync(Connection.java:290)
	at com.rethinkdb.net.Connection.noreplyWait(Connection.java:298)
	at com.rethinkdb.net.Connection.close(Connection.java:356)
	at com.rethinkdb.net.Connection.close(Connection.java:344)
	at e.i.DB.getConnection(DB.java:60)
...
15:00:45:192 [whois-server] [ERROR] WhoisServer - Processing failed: Client already connected!
com.rethinkdb.gen.exc.ReqlDriverError: Client already connected!
	at com.rethinkdb.net.Connection.connectAsync(Connection.java:127)
	at com.rethinkdb.net.Connection.connect(Connection.java:144)
	at e.i.DB.getConnection(DB.java:64)

And the getConnection is in essence:

    static Connection getConnection() {
         Connection conn = sessionThreadLocal.get();
        if (conn.isOpen()) {
             return conn;
        } else {
             logger.info("Reconnecting to DB");
            try {
                conn.close();
            } catch (ReqlDriverError e) {
                logger.info("Closing failed, ignoring", e);
            }
            return conn.connect();
         }
     }

isOpen is false, close fails because pump is closed (whatever that means in laymans terms) and connecting again is not possible.

This is pretty grave bug which renders the DB almost unusable from Java (or lacks guidance on how usage should be orchestrated under normal circumstances).

I could provide a docker setup but this seems easily repeatable, given just the undefined wait time of 15 or so minutes.

[NotJustAnna]

Version 2.4.4 will be released soon. It may not contain a fix (yet), but it should add a better shutdown reason. Which should allow me to get a better understanding of what's going on.

[NotJustAnna]

In answer to @martinpaljak:

The response pump is what basically links a ReqlAst.run(conn) to the Result<T>. I'm not the one who invented it, and it basically reads from the TCP socket and feed awaiting queries. If the response pump is shut down, this means either the socket closed or an exception happened.

I don't know why, but you're the only two people that ever reported this. I know other users with no problems at all regarding this. I hope that v2.4.4 will expose what's happening.

[martinpaljak]

@adriantodt Thanks! I'll try this PR ASAP to see what it throws.

[martinpaljak]

java.net.SocketException: Connection reset
com.rethinkdb.gen.exc.ReqlDriverError: java.net.SocketException: Connection reset
	at com.rethinkdb.net.DefaultConnectionFactory$SocketWrapper.read(DefaultConnectionFactory.java:153)
	at com.rethinkdb.net.Response.readFromSocket(Response.java:88)
	at com.rethinkdb.net.DefaultConnectionFactory$ThreadResponsePump.lambda$new$1(DefaultConnectionFactory.java:201)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.net.SocketException: Connection reset
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:186)
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:140)
	at com.rethinkdb.net.DefaultConnectionFactory$SocketWrapper.read(DefaultConnectionFactory.java:144)
	... 3 more
com.rethinkdb.gen.exc.ReqlDriverError: java.net.SocketException: Connection reset
	at com.rethinkdb.net.DefaultConnectionFactory$SocketWrapper.read(DefaultConnectionFactory.java:153)
	at com.rethinkdb.net.Response.readFromSocket(Response.java:88)
	at com.rethinkdb.net.DefaultConnectionFactory$ThreadResponsePump.lambda$new$1(DefaultConnectionFactory.java:201)
	at java.base/java.lang.Thread.run(Thread.java:834)
Caused by: java.net.SocketException: Connection reset
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:186)
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:140)
	at com.rethinkdb.net.DefaultConnectionFactory$SocketWrapper.read(DefaultConnectionFactory.java:144)
	... 3 more
java.net.SocketException: Connection reset
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:186)
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:140)
	at com.rethinkdb.net.DefaultConnectionFactory$SocketWrapper.read(DefaultConnectionFactory.java:144)
	at com.rethinkdb.net.Response.readFromSocket(Response.java:88)
	at com.rethinkdb.net.DefaultConnectionFactory$ThreadResponsePump.lambda$new$1(DefaultConnectionFactory.java:201)
	at java.base/java.lang.Thread.run(Thread.java:834)

With what is on maven central as 2.4.4

[NotJustAnna]

That's a server issue, not a client issue.

[martinpaljak]

Why so? Shouldn’t the client recover or provide a way to reconnect at least?

[martinpaljak]

FYI: this is a standard dockerized deployment with both rethink and the client application on the same physical host connecting via overlay network. Before this exception it used to be the "pump closed" one. Sorry for a repeated stacktrace.