@Test public void testSocketTimeout() { Configuration clientConfig = getBlobClientConfig(); int oldSoTimeout = clientConfig.getInteger(BlobServerOptions.SO_TIMEOUT); clientConfig.setInteger(BlobServerOptions.SO_TIMEOUT, 50); getBlobServer().setBlockingMillis(10_000); try { InetSocketAddress serverAddress = new InetSocketAddress("localhost", getBlobServer().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(), is(true)); } } finally { clientConfig.setInteger(BlobServerOptions.SO_TIMEOUT, oldSoTimeout); getBlobServer().setBlockingMillis(0); } }
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(); 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); } }
BlobClient implements Closeable { 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(); 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); } } }
BlobClient implements Closeable { 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(); 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); } } BlobClient(InetSocketAddress serverAddress, Configuration clientConfig); }
BlobClient implements Closeable { 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(); 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); } } BlobClient(InetSocketAddress serverAddress, Configuration clientConfig); @Override void close(); boolean isClosed(); static List<PermanentBlobKey> uploadFiles(
			InetSocketAddress serverAddress, Configuration clientConfig, JobID jobId, List<Path> files); PermanentBlobKey uploadFile(JobID jobId, Path file); }
BlobClient implements Closeable { 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(); 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); } } BlobClient(InetSocketAddress serverAddress, Configuration clientConfig); @Override void close(); boolean isClosed(); static List<PermanentBlobKey> uploadFiles(
			InetSocketAddress serverAddress, Configuration clientConfig, JobID jobId, List<Path> files); PermanentBlobKey uploadFile(JobID jobId, Path file);  }