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.

[NotJustAnna]

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

There's no code related to connection recovering, and I think what was closer to it was the ConnectionPool, which someone else removed way before I got on the project.

As far as I know there's no "connection resuming" on RethinkDB (citation needed), I wouldn't mind making "connection recovering" a thing, but that would be a LOT of effort, but doable.

[Kodehawa]

Check your language is UTF-8 (on linux, check if localectl says LANG=en_US.UTF-8 or any variation ending with .UTF-8). There has been issues with this before.

(This on where the java client is running, not where rtdb is)

[NotJustAnna]

Apparently Docker have issues with TCP KeepAlive connections (see moby/moby#31208). You all mention "15/20 minutes", which is how long this bug also reports.

[martinpaljak]

@adriantodt thanks for the pointer! I would not be surprised if docker brings out the best of weirdnesses (just had a 6 hour headbang/wtf debugging side effects of a meaningless configuration line). But I would also expect a robust client for any kind of networked thing to recover from various types of bad network cases, reliably. For example, providing some kind of application level keepalive? What made me scratch my head was trying to do a "if error, close and open again" and that is not working, just re-opening a fresh connection seems to do the trick.

[doantrungvn]

Check your language is UTF-8 (on linux, check if localectl says LANG=en_US.UTF-8 or any variation ending with .UTF-8). There has been issues with this before.

(This on where the java client is running, not where rtdb is)

@Kodehawa Ko
What's wrong with locale language? UTF-8 is wrong or not? If UTF-8 is wrong, how to set language to fix my problem?
Pls explain to me?

[doantrungvn]

@adriantodt @martinpaljak
I have subscribe to a feed by calling changes on a table (changfeed).
If I cannot control the Response pump close error or the connection error then changefeed is meaningless (don't receive changes on a table when have connection error)
Re-opening a fresh connection seems to do the trick but not effective with my case.

[martinpaljak]

@adriantodt Reducing keepalive time in the client container seems to help. But it would be nice and reassuring if the client could elegantly recover from network element issues between itself and a server. Either using application-level keepalives or providing a sane "connection reset, reconnecting" kind of flow automatically or manually via API. Or document that a connection is single-use and can't be re-used once closed (the isOpen() method did detect it properly earlier)