InputStream getInternal(@Nullable JobID jobId, BlobKey blobKey) throws IOException {

        if (this.socket.isClosed()) {
            throw new IllegalStateException(
                    "BLOB Client is not connected. "
                            + "Client has been shut down or encountered an error before.");
        }
        if (LOG.isDebugEnabled()) {
            LOG.debug("GET BLOB {}/{} from {}.", jobId, blobKey, socket.getLocalSocketAddress());
        }

        try {
            OutputStream os = this.socket.getOutputStream();
            InputStream is = this.socket.getInputStream();

            // Send GET header
            sendGetHeader(os, jobId, blobKey);
            receiveAndCheckGetResponse(is);

            return new BlobInputStream(is, blobKey, os);
        } catch (Throwable t) {
            BlobUtils.closeSilently(socket, LOG);
            throw new IOException("GET operation failed: " + t.getMessage(), t);
        }
    }
@Test
    void testSocketTimeout() throws IOException {
        Configuration clientConfig = getBlobClientConfig();
        int oldSoTimeout = clientConfig.get(BlobServerOptions.SO_TIMEOUT);

        clientConfig.set(BlobServerOptions.SO_TIMEOUT, 50);

        try (final TestBlobServer testBlobServer =
                new TestBlobServer(
                        clientConfig,
                        tempDir.resolve("test_server").toFile(),
                        new VoidBlobStore(),
                        10_000L)) {
            testBlobServer.start();
            InetSocketAddress serverAddress =
                    new InetSocketAddress("localhost", testBlobServer.getPort());

            try (BlobClient client = new BlobClient(serverAddress, clientConfig)) {
                client.getInternal(new JobID(), BlobKey.createKey(TRANSIENT_BLOB));

                fail("Should throw an exception.");
            } catch (Throwable t) {
                assertThat(ExceptionUtils.findThrowable(t, java.net.SocketTimeoutException.class))
                        .isPresent();
            }
        } finally {
            clientConfig.set(BlobServerOptions.SO_TIMEOUT, oldSoTimeout);
        }
    }