openChannel can block forever on channel.getInputStream().peek()

[subdir]

We want to be able to terminate our program early (e.g. on SIGTERM).

It was possible to some extent before this commit 8dd49eb

You could just call BinaryLogClient.disconnect() on program termination. If BinaryLogClient.channel is already set, then it disconnects, if not - no big deal (that being said, there were possible race conditions).

Now disconnect waits for connectLock which is released only after channel is created and at least one byte is successfully read. The problem is that in theory channel.peek() may block for indefinitely long time, so we have no way to shutdown gracefully.

Possible solution is to introduce connect timeout. It may be used not only with peek but with socket.connect as well.

[shyiko]

Hi @subdir.

I'm not sure that can actually happen. BinaryLogClient.disconnect() is trying to close the connection in a while loop over the connectLock (https://github.com/shyiko/mysql-binlog-connector-java/blob/master/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java#L988). Now, if there is a peek() that is blocked it's going to throw SocketException as soon as channel.close() is called (https://github.com/shyiko/mysql-binlog-connector-java/blob/master/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java#L1004) (it's calling socket.close() under the hood (among other things)). So, unless I'm missing something, the situation you described should never occur (it would be a major disaster if it could).

Let me know if I got it wrong.

[shyiko]

Alright, feel free to reopen the ticket if there is a problem after all.

[snaury]

@shyiko I'm not the one who reported this, though I was the one who initially suspected there might be an issue and told @subdir about it. The problem is that channel.close will never be called, because channel is assigned only when openChannel returns, however it only returns after peek succeeds.

Now imagine that you accidentally connect to a port that will never reply first, like some-host:80 for http. In that case the following will happen:

• The connectLock is held at least until peek in openChannel returns
• The channel is null until openChannel returns, so it's impossible to close it
• There are no configurable timeouts in BinaryLogClient for sockets, so peek may never return until the server gives up and closes the connection
• The disconnect method will block until connectLock is released, which may never happen even under normal circumstances (networks are not reliable, firewalls may start dropping packets at any time, etc.)

Before openChannel was extracted it wasn't so bad, because the only place this could happen was during a socket.connect and OSes have an implicit connect timeout of 30 seconds, so worst case would have been a 30 second delay, which could even be avoided by moving everything to a separate daemon thread, calling disconnect and then forgetting about that thread even if it temporarily stuck in a connect call.

Now it became very dangerous, because even if we move everything to a separate thread and insulate ourselves as much as possible, then disconnect may be stuck forever anyway, and the only choice would be to move disconnect to a separate thread as well. :(

I think there should a safe asynchronous variant of disconnect which wouldn't spin forever when channel == null, but which would return after setting a disconnection flag (to be be checked before peek is called). Probably even better is for socket.close to be called somehow if it's inside socket.connect too. However interaction with keepalive thread might complicate things severely.

Also, have you though about making it possible to set some read/write timeouts maybe?

[shyiko]

The problem is that channel.close will never be called, because channel is assigned only when openChannel returns, however it only returns after peek succeeds.

I can't believe I missed it 🤦‍♂️ Thank you for pointing it out. I'll try to release a hotfix ASAP.

As for the timeouts, "connect" timeout is automatically set if BinaryLogClient.connect(timeout) is used (since 0.9.0) and can be manually set via BinaryLogClient::setConnectTimeout.
Unfortunately, changing read/write timeouts isn't as simple.
While read timeout can be set with

binaryLogClient.setSocketFactory(() -> {
    Socket socket = new Socket();
    socket.setSoTimeout((int) TimeUnit.SECONDS.toMillis(30));
    return socket;
});

there is no way to set write timeout (http://bugs.java.com/bugdatabase/view_bug.do?bug_id=4031100). ...unless of course you are willing to add org.apache.hadoop:hadoop-common:2.6.5 as a dependency, which contains special SocketInputStream/SocketOutputStream (see example below).

import org.apache.hadoop.net.SocketInputStream;
import org.apache.hadoop.net.SocketOutputStream;

...

binaryLogClient.setSocketFactory(new SocketFactory() {
    @Override
    public Socket createSocket() throws SocketException {
        class LazyInitInputStream extends FilterInputStream {
            LazyInitInputStream() { super(null); }
            void init(InputStream in) {
                this.in = in;
            }
        }
        class LazyInitOutputStream extends FilterOutputStream {
            LazyInitOutputStream() { super(null); }
            void init(OutputStream out) {
                this.out = out;
            }
        }
        return new Socket() {

            Socket socket;
            LazyInitInputStream in = new LazyInitInputStream();
            LazyInitOutputStream out = new LazyInitOutputStream();

            @Override
            public InputStream getInputStream() throws IOException {
                return in;
            }
            @Override
            public OutputStream getOutputStream() throws IOException {
                return out;
            }

            @Override
            public void connect(SocketAddress endpoint, int timeout) 
                    throws IOException {
                this.socket = SocketChannel.open().socket();
                this.socket.connect(endpoint, timeout);
                in.init(new SocketInputStream(this.socket, 
                    TimeUnit.SECONDS.toMillis(3)));
                out.init(new SocketOutputStream(this.socket, 
                    TimeUnit.SECONDS.toMillis(3)));
            }

            @Override
            public boolean isClosed() {
                return this.socket.isClosed();
            }

            @Override
            public void shutdownInput() throws IOException {
                this.socket.shutdownInput();
            }

            @Override
            public void shutdownOutput() throws IOException {
                this.socket.shutdownOutput();
            }

            @Override
            public void close() throws IOException {
                this.socket.close();
            }
        };
    }
});

I was planning to switch to SocketChannel a long time ago but honestly I just haven't found the time for it.

Now, normally you would not care about read/write timeout and would rely on "keepAlive" thread to monitor the connection. Unfortunately, as you already pointed out in #118, current implementation is fundamentally flawed. I think it's time I fix it.

tldr; I'm releasing a fix for this issue right away. #118 will be fixed within the next few days. It should free you from mangling with read/write timeouts. You would only need to specify connect timeout and keepAlive strategy.

Sounds good?

[snaury]

Thanks! Regarding #118 it is no longer a priority, since we're already managing keepalive ourselves, i.e. we are wrapping BinaryLogClient in an asynchronous queue, and when there's a timeout reading from that queue we simply abandon the old client and create a new one. It has advantages too, since we keep track of a committed gtid set and it's guaranteed to work when our masters switch (I'm not sure how BinaryLogClient would handle a case when reading failed mid-transaction and a new master didn't have that gtid yet, right now I'm emitting a synthetic rollback event in that case, but it's not a generic solution).

[snaury]

Oh, and thank you very much regarding setSocketFactory example, that's going to be very handy!

[shyiko]

0.9.2 deployed to Maven Central 🎉 (sync might take up to ~20 minutes).

Thank you, @subdir and @snaury!

[subdir]

Oh, looks like I am too late. Sorry. Thank you guys for taking care of it.