chore: log zero bytes read

src/main/java/com/google/cloud/spanner/pgadapter/NonBlockingProxyServer.java

@@ -56,22 +56,25 @@
             new InetSocketAddress((InetAddress) null, getLocalPort()),
             ++index);
     if (getLocalPort() == 0) {
       this.localPort = tcpListener.getServerSocketChannel().socket().getLocalPort();
     }
     listenersBuilder.add(tcpListener);
 
     if (optionsMetadata.isDomainSocketEnabled()) {
-      File tempDir = new File(optionsMetadata.getSocketFile(getLocalPort()));
-      if (tempDir.getParentFile() != null && !tempDir.getParentFile().exists()) {
+      File socketFile = new File(optionsMetadata.getSocketFile(getLocalPort()));
+      if (socketFile.getParentFile() != null && !socketFile.getParentFile().exists()) {
         //noinspection ResultOfMethodCallIgnored
-        tempDir.getParentFile().mkdirs();
+        socketFile.getParentFile().mkdirs();
+      }
+      if (socketFile.exists() && !socketFile.delete()) {
+        throw new IOException("Failed to re-create socket file");
       }
       NonBlockingServerListener listener =
           createServerListener(
               AFUNIXSelectorProvider.provider().openSelector(),
               AFUNIXSelectorProvider.provider().openSelector(),
               AFUNIXServerSocketChannel.open(),
-              AFUNIXSocketAddress.of(tempDir),
+              AFUNIXSocketAddress.of(socketFile),
               ++index);
       listenersBuilder.add(listener);
     }
@@ -107,9 +110,9 @@
       }
       notifyStarted();
       logger.log(Level.INFO, "Non-blocking server started");
     } catch (Throwable throwable) {
       logger.log(Level.WARNING, "Non-blocking server failed to start", throwable);
       notifyFailed(throwable);
     }
   }
 
@@ -119,8 +122,8 @@
       try {
         listener.getAcceptSelector().close();
         logger.log(Level.INFO, "Closed listening selector {}", listener.getAcceptSelector());
       } catch (IOException ioException) {
         logger.log(Level.WARNING, "Failed to close selector", ioException);
       }
     }
     for (ConnectionHandler handler : getConnectionHandlers()) {
@@ -129,14 +132,14 @@
     logger.log(Level.INFO, "Terminated all active connections");
     try {
       SpannerPool.closeSpannerPool();
     } catch (Throwable ignore) {
     }
     for (NonBlockingServerListener listener : serverListeners) {
       try {
         listener.getServerSocketChannel().close();
         logger.log(Level.INFO, "Closed listening socket {}", listener.getServerSocketChannel());
       } catch (IOException ioException) {
         logger.log(Level.WARNING, "Failed to close socket", ioException);
       }
     }
     notifyStopped();

src/main/java/com/google/cloud/spanner/pgadapter/NonBlockingSocketReader.java

@@ -55,9 +55,9 @@
         new Listener() {
           @Override
           public void failed(State from, Throwable failure) {
             super.failed(from, failure);
             running.set(false);
           }
 
           @Override
           public void terminated(State from) {
@@ -92,8 +92,8 @@
           keys.clear();
         }
       }
     } catch (IOException ioException) {
       logger.log(Level.WARNING, "selectNow for reader failed", ioException);
     }
   }
 
@@ -104,15 +104,15 @@
     }
     SocketChannel channel = (SocketChannel) key.channel();
     if (!(channel.isOpen() && channel.isConnected())) {
       return;
     }
 
     try {
       if (handler.getStatus() == ConnectionStatus.UNAUTHENTICATED) {
-        ByteBuffer lengthBuffer = read(4, channel);
+        ByteBuffer lengthBuffer = read(4, channel, true);
         lengthBuffer.rewind();
         int length = ByteConverter.int4(lengthBuffer.array(), 0);
-        ByteBuffer dataBuffer = read(length, lengthBuffer, channel);
+        ByteBuffer dataBuffer = read(length, lengthBuffer, channel, true);
         dataBuffer.rewind();
         try (ByteBufferInputStream inputStream = new ByteBufferInputStream(dataBuffer)) {
           handler.setRawInputStream(inputStream);
@@ -121,13 +121,14 @@
         }
       } else {
         // All control messages has a 1-byte type + 4 byte length.
-        ByteBuffer headerBuffer = read(handler.getHeaderBuffer(), channel);
+        ByteBuffer headerBuffer = read(handler.getHeaderBuffer(), channel, false);
         byte[] dst = new byte[4];
         headerBuffer.position(1);
         headerBuffer.get(dst);
         int length = ByteConverter.int4(dst, 0);
         headerBuffer.rewind();
-        ByteBuffer message = read(handler.getMessageBuffer(length + 1), headerBuffer, channel);
+        ByteBuffer message =
+            read(handler.getMessageBuffer(length + 1), headerBuffer, channel, false);
         message.rewind();
         try (ByteBufferInputStream inputStream = new ByteBufferInputStream(message)) {
           handler.setRawInputStream(inputStream);
@@ -154,28 +155,39 @@
     }
   }
 
-  private static ByteBuffer read(int length, SocketChannel channel) throws IOException {
-    return read(length, null, channel);
+  private static ByteBuffer read(int length, SocketChannel channel, boolean bootstrap)
+      throws IOException {
+    return read(length, null, channel, bootstrap);
   }
 
-  private static ByteBuffer read(ByteBuffer destination, SocketChannel channel) throws IOException {
-    return read(destination, null, channel);
+  private static ByteBuffer read(ByteBuffer destination, SocketChannel channel, boolean bootstrap)
+      throws IOException {
+    return read(destination, null, channel, bootstrap);
   }
 
-  private static ByteBuffer read(int length, ByteBuffer header, SocketChannel channel)
-      throws IOException {
+  private static ByteBuffer read(
+      int length, ByteBuffer header, SocketChannel channel, boolean bootstrap) throws IOException {
     ByteBuffer destination = ByteBuffer.allocate(length);
-    return read(destination, header, channel);
+    return read(destination, header, channel, bootstrap);
   }
 
-  private static ByteBuffer read(ByteBuffer destination, ByteBuffer header, SocketChannel channel)
+  private static ByteBuffer read(
+      ByteBuffer destination, ByteBuffer header, SocketChannel channel, boolean bootstrap)
       throws IOException {
     if (header != null) {
       destination.put(header);
     }
-    int read;
+    int read, zeroBytesCounter = 0;
     do {
       read = channel.read(destination);
+      if (read == 0) {
+        zeroBytesCounter++;
+        if (zeroBytesCounter % 1000 == 0) {
+          System.out.println("Read zero bytes " + zeroBytesCounter + " times");
+          System.out.println("Expecting " + destination.capacity() + " bytes");
+          System.out.println("Remaining " + destination.remaining() + " bytes");
+        }
+      }
     } while (read > -1 && destination.hasRemaining());
     if (read == -1) {
       throw new EOFException();

src/test/java/com/google/cloud/spanner/pgadapter/python/psycopg2/PythonTransactionTests.java

@@ -22,7 +22,6 @@
 import com.google.cloud.spanner.Statement;
 import com.google.cloud.spanner.pgadapter.python.PythonTest;
 import com.google.cloud.spanner.pgadapter.wireprotocol.QueryMessage;
-import com.google.cloud.spanner.pgadapter.wireprotocol.WireMessage;
 import com.google.common.collect.ImmutableList;
 import com.google.protobuf.AbstractMessage;
 import com.google.protobuf.ByteString;

src/test/java/com/google/cloud/spanner/pgadapter/python/sqlalchemy/SqlAlchemyOrmTest.java

@@ -54,7 +54,7 @@ public class SqlAlchemyOrmTest extends AbstractMockServerTest {
 
   @Parameters(name = "host = {0}")
   public static List<Object[]> data() {
-    return ImmutableList.of(new Object[] {"localhost"}/*, new Object[] {""}*/);
+    return ImmutableList.of(new Object[] {"localhost"} /*, new Object[] {""}*/);
   }
 
   @BeforeClass