Update usages of DefaultBlockWorkerClient

core/client/src/main/java/alluxio/client/block/AlluxioBlockStore.java

@@ -232,7 +232,7 @@ public void promote(long blockId) throws IOException {
     // Get the first worker address for now, as this will likely be the location being read from
     // TODO(calvin): Get this location via a policy (possibly location is a parameter to promote)
     WorkerNetAddress workerAddr = info.getLocations().get(0).getWorkerAddress();
-    DefaultBlockWorkerClient blockWorkerClient = mContext.acquireWorkerClient(workerAddr);
+    BlockWorkerClient blockWorkerClient = mContext.acquireWorkerClient(workerAddr);
     try {
       blockWorkerClient.promoteBlock(blockId);
     } catch (AlluxioException e) {

core/client/src/main/java/alluxio/client/block/BlockStoreContext.java

@@ -128,11 +128,11 @@ public void close() {
    * Obtains a client for a worker with the given address.
    *
    * @param address the address of the worker to get a client to
-   * @return a {@link DefaultBlockWorkerClient} connected to the worker with the given hostname
+   * @return a {@link BlockWorkerClient} connected to the worker with the given hostname
    * @throws IOException if no Alluxio worker is available for the given hostname
    */
-  public DefaultBlockWorkerClient acquireWorkerClient(WorkerNetAddress address) throws IOException {
-    DefaultBlockWorkerClient client;
+  public BlockWorkerClient acquireWorkerClient(WorkerNetAddress address) throws IOException {
+    BlockWorkerClient client;
     if (address == null) {
       throw new RuntimeException(ExceptionMessage.NO_WORKER_AVAILABLE.getMessage());
     }
@@ -151,9 +151,9 @@ public DefaultBlockWorkerClient acquireWorkerClient(WorkerNetAddress address) th
   /**
    * Obtains a worker client on the local worker in the system. For testing only.
    *
-   * @return a {@link DefaultBlockWorkerClient} to a worker in the Alluxio system or null if failed
+   * @return a {@link BlockWorkerClient} to a worker in the Alluxio system or null if failed
    */
-  public DefaultBlockWorkerClient acquireLocalWorkerClient() {
+  public BlockWorkerClient acquireLocalWorkerClient() {
     initializeLocalBlockWorkerClientPool();
     if (mLocalBlockWorkerClientPoolMap.isEmpty()) {
       return null;
@@ -167,10 +167,10 @@ public DefaultBlockWorkerClient acquireLocalWorkerClient() {
    *
    * @param address worker address
    *
-   * @return a {@link DefaultBlockWorkerClient} to the given worker address or null if no such worker can
+   * @return a {@link BlockWorkerClient} to the given worker address or null if no such worker can
    *         be found
    */
-  public DefaultBlockWorkerClient acquireLocalWorkerClient(WorkerNetAddress address) {
+  public BlockWorkerClient acquireLocalWorkerClient(WorkerNetAddress address) {
     initializeLocalBlockWorkerClientPool();
     if (!mLocalBlockWorkerClientPoolMap.containsKey(address)) {
       return null;
@@ -186,7 +186,7 @@ public DefaultBlockWorkerClient acquireLocalWorkerClient(WorkerNetAddress addres
    * @param address the address of the worker
    * @return a worker client with a connection to the specified hostname
    */
-  private DefaultBlockWorkerClient acquireRemoteWorkerClient(WorkerNetAddress address) {
+  private BlockWorkerClient acquireRemoteWorkerClient(WorkerNetAddress address) {
     // If we couldn't find a worker, crash.
     if (address == null) {
       // TODO(calvin): Better exception usage.
@@ -200,13 +200,13 @@ private DefaultBlockWorkerClient acquireRemoteWorkerClient(WorkerNetAddress addr
   }
 
   /**
-   * Releases the {@link DefaultBlockWorkerClient} back to the client pool, or destroys it if it was a
+   * Releases the {@link BlockWorkerClient} back to the client pool, or destroys it if it was a
    * remote client.
    *
    * @param blockWorkerClient the worker client to release, the client should not be accessed after
    *        this method is called
    */
-  public void releaseWorkerClient(DefaultBlockWorkerClient blockWorkerClient) {
+  public void releaseWorkerClient(BlockWorkerClient blockWorkerClient) {
     // If the client is local and the pool exists, release the client to the pool, otherwise just
     // close the client.
     if (blockWorkerClient.isLocal()) {

core/client/src/main/java/alluxio/client/block/BlockWorkerClientHeartbeatExecutor.java

@@ -23,9 +23,9 @@
  */
 @ThreadSafe
 final class BlockWorkerClientHeartbeatExecutor implements HeartbeatExecutor {
-  private final DefaultBlockWorkerClient mBlockWorkerClient;
+  private final BlockWorkerClient mBlockWorkerClient;
 
-  public BlockWorkerClientHeartbeatExecutor(DefaultBlockWorkerClient blockWorkerClient) {
+  public BlockWorkerClientHeartbeatExecutor(BlockWorkerClient blockWorkerClient) {
     mBlockWorkerClient = Preconditions.checkNotNull(blockWorkerClient);
   }
 

core/client/src/main/java/alluxio/client/block/BlockWorkerClientPool.java

@@ -29,7 +29,7 @@
  * thread is done using the client.
  */
 @ThreadSafe
-final class BlockWorkerClientPool extends ResourcePool<DefaultBlockWorkerClient> {
+final class BlockWorkerClientPool extends ResourcePool<BlockWorkerClient> {
   private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
   /**
    * The capacity for this pool must be large, since each block written will hold a client until
@@ -53,7 +53,7 @@ public void close() {
   }
 
   @Override
-  public void release(DefaultBlockWorkerClient blockWorkerClient) {
+  public void release(BlockWorkerClient blockWorkerClient) {
     try {
       // Heartbeat to send the client metrics.
       blockWorkerClient.sessionHeartbeat();
@@ -65,7 +65,7 @@ public void release(DefaultBlockWorkerClient blockWorkerClient) {
   }
 
   @Override
-  protected DefaultBlockWorkerClient createNewResource() {
+  protected BlockWorkerClient createNewResource() {
     long clientId = IdUtils.getRandomNonNegativeLong();
     return new DefaultBlockWorkerClient(mWorkerNetAddress, ClientContext.getBlockClientExecutorService(),
         clientId, true, ClientContext.getClientMetrics());

core/client/src/main/java/alluxio/client/block/LocalBlockInStream.java

@@ -35,7 +35,7 @@ public final class LocalBlockInStream extends BufferedBlockInStream {
   /** Helper to manage closables. */
   private final Closer mCloser;
   /** Client to communicate with the local worker. */
-  private final DefaultBlockWorkerClient mBlockWorkerClient;
+  private final BlockWorkerClient mBlockWorkerClient;
   /** The block store context which provides block worker clients. */
   private final BlockStoreContext mContext;
   /** The file reader to read a local block. */

core/client/src/main/java/alluxio/client/block/LocalBlockOutStream.java

@@ -37,7 +37,7 @@
 public final class LocalBlockOutStream extends BufferedBlockOutStream {
   private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
   private final Closer mCloser;
-  private final DefaultBlockWorkerClient mBlockWorkerClient;
+  private final BlockWorkerClient mBlockWorkerClient;
   private final LocalFileBlockWriter mWriter;
   private long mReservedBytes;
 

core/client/src/main/java/alluxio/client/block/RemoteBlockInStream.java

@@ -38,7 +38,7 @@ public final class RemoteBlockInStream extends BufferedBlockInStream {
   private final Long mLockId;
 
   /** Client to communicate with the remote worker. */
-  private final DefaultBlockWorkerClient mBlockWorkerClient;
+  private final BlockWorkerClient mBlockWorkerClient;
   /** The block store context which provides block worker clients. */
   private final BlockStoreContext mContext;
   private final ClientMetrics mMetrics;

core/client/src/main/java/alluxio/client/block/RemoteBlockOutStream.java

@@ -27,7 +27,7 @@
 @NotThreadSafe
 public final class RemoteBlockOutStream extends BufferedBlockOutStream {
   private final RemoteBlockWriter mRemoteWriter;
-  private final DefaultBlockWorkerClient mBlockWorkerClient;
+  private final BlockWorkerClient mBlockWorkerClient;
   private final ClientMetrics mMetrics;
 
   /**

core/client/src/test/java/alluxio/client/block/AlluxioBlockStoreTest.java

@@ -35,7 +35,7 @@
  * Tests for {@link AlluxioBlockStore}.
  */
 @RunWith(PowerMockRunner.class)
-@PrepareForTest({BlockStoreContext.class, DefaultBlockWorkerClient.class})
+@PrepareForTest({BlockStoreContext.class})
 public final class AlluxioBlockStoreTest {
   private static final long BLOCK_ID = 3L;
   private static final long BLOCK_LENGTH = 1000L;
@@ -72,11 +72,11 @@ public final class AlluxioBlockStoreTest {
   public TemporaryFolder mTestFolder = new TemporaryFolder();
 
   private BlockMasterClient mMasterClient;
-  private DefaultBlockWorkerClient mBlockWorkerClient;
+  private BlockWorkerClient mBlockWorkerClient;
 
   @Before
   public void before() throws Exception {
-    mBlockWorkerClient = PowerMockito.mock(DefaultBlockWorkerClient.class);
+    mBlockWorkerClient = PowerMockito.mock(BlockWorkerClient.class);
     mMasterClient = PowerMockito.mock(BlockMasterClient.class);
   }
 

core/client/src/test/java/alluxio/client/block/BlockStoreContextTest.java

@@ -128,7 +128,7 @@ public void acquireWorkerLimitTest() throws Exception {
         .withArguments(Mockito.any(), Mockito.any(), Mockito.anyLong(), Mockito.anyBoolean(),
             Mockito.any()).thenReturn(workerClientMock);
 
-    final List<DefaultBlockWorkerClient> clients = new ArrayList<>();
+    final List<BlockWorkerClient> clients = new ArrayList<>();
 
     // Reduce the size of the worker thread pool to lower the chance of a timeout.
     Configuration.set(Constants.USER_BLOCK_WORKER_CLIENT_THREADS, "10");
@@ -154,7 +154,7 @@ public void acquireWorkerLimitTest() throws Exception {
     // Release all the clients
     // Set the RPC number of retries to -1 to prevent the worker client from trying to send a
     // heartbeat message when it is released.
-    for (DefaultBlockWorkerClient client : clients) {
+    for (BlockWorkerClient client : clients) {
       BlockStoreContext.INSTANCE.releaseWorkerClient(client);
     }
 
@@ -210,7 +210,7 @@ public void hasNoLocalWorkerTest() throws Exception {
   class AcquireWorkerClient implements Runnable {
     @Override
     public void run() {
-      DefaultBlockWorkerClient client = BlockStoreContext.INSTANCE.acquireLocalWorkerClient();
+      BlockWorkerClient client = BlockStoreContext.INSTANCE.acquireLocalWorkerClient();
       BlockStoreContext.INSTANCE.releaseWorkerClient(client);
     }
   }

core/client/src/test/java/alluxio/client/block/BlockWorkerClientHeartbeatExecutorTest.java

@@ -12,25 +12,19 @@
 package alluxio.client.block;
 
 import org.junit.Test;
-import org.junit.runner.RunWith;
 import org.mockito.Mockito;
-import org.powermock.api.mockito.PowerMockito;
-import org.powermock.core.classloader.annotations.PrepareForTest;
-import org.powermock.modules.junit4.PowerMockRunner;
 
 /**
  * Tests for the {@link BlockWorkerClientHeartbeatExecutor} class.
  */
-@RunWith(PowerMockRunner.class)
-@PrepareForTest(DefaultBlockWorkerClient.class)
 public class BlockWorkerClientHeartbeatExecutorTest {
 
   /**
    * Tests to ensure heartbeat calls BlockWorkerClient.periodicHeartbeat.
    */
   @Test
   public void heartbeatCallsPeriodicHeartbeat() throws Exception {
-    DefaultBlockWorkerClient mock = PowerMockito.mock(DefaultBlockWorkerClient.class);
+    BlockWorkerClient mock = Mockito.mock(BlockWorkerClient.class);
     BlockWorkerClientHeartbeatExecutor heartbeatExecutor =
         new BlockWorkerClientHeartbeatExecutor(mock);
 

core/client/src/test/java/alluxio/client/file/FileOutStreamTest.java

@@ -18,7 +18,6 @@
 import alluxio.client.WriteType;
 import alluxio.client.block.AlluxioBlockStore;
 import alluxio.client.block.BlockStoreContext;
-import alluxio.client.block.DefaultBlockWorkerClient;
 import alluxio.client.block.BlockWorkerInfo;
 import alluxio.client.block.BufferedBlockOutStream;
 import alluxio.client.block.TestBufferedBlockOutStream;
@@ -65,7 +64,7 @@
  */
 @RunWith(PowerMockRunner.class)
 @PrepareForTest({FileSystemContext.class, BlockStoreContext.class, FileSystemMasterClient.class,
-    AlluxioBlockStore.class, UnderFileSystem.class, DefaultBlockWorkerClient.class})
+    AlluxioBlockStore.class, UnderFileSystem.class})
 public class FileOutStreamTest {
 
   private static final long BLOCK_LENGTH = 100L;

core/common/src/main/java/alluxio/Client.java

@@ -40,4 +40,10 @@ public interface Client extends Closeable {
    * @return true if this client is connected to the remote
    */
   boolean isConnected();
+
+  /**
+   * Closes the connection with the remote permanently. This instance should be not be reused after
+   * closing.
+   */
+  void close();
 }

tests/src/test/java/alluxio/master/ServiceSocketBindIntegrationTest.java

@@ -16,7 +16,7 @@
 import alluxio.LocalAlluxioClusterResource;
 import alluxio.client.block.BlockMasterClient;
 import alluxio.client.block.BlockStoreContext;
-import alluxio.client.block.DefaultBlockWorkerClient;
+import alluxio.client.block.BlockWorkerClient;
 import alluxio.client.block.RetryHandlingBlockMasterClient;
 import alluxio.exception.ConnectionFailedException;
 import alluxio.util.network.NetworkAddressUtils;
@@ -45,7 +45,7 @@ public class ServiceSocketBindIntegrationTest {
 
   private BlockMasterClient mBlockMasterClient;
   private HttpURLConnection mMasterWebService;
-  private DefaultBlockWorkerClient mBlockWorkerClient;
+  private BlockWorkerClient mBlockWorkerClient;
   private SocketChannel mWorkerDataService;
   private HttpURLConnection mWorkerWebService;
 

tests/src/test/java/alluxio/security/BlockWorkerClientAuthenticationIntegrationTest.java

@@ -13,6 +13,7 @@
 
 import alluxio.Constants;
 import alluxio.LocalAlluxioClusterResource;
+import alluxio.client.block.BlockWorkerClient;
 import alluxio.client.block.DefaultBlockWorkerClient;
 import alluxio.client.util.ClientTestUtils;
 import alluxio.security.MasterClientAuthenticationIntegrationTest.NameMatchAuthenticationProvider;
@@ -88,7 +89,7 @@ public void customAuthenticationDenyConnectTest() throws Exception {
     mThrown.expect(IOException.class);
     mThrown.expectMessage("Failed to connect to the worker");
 
-    try (DefaultBlockWorkerClient blockWorkerClient = new DefaultBlockWorkerClient(
+    try (BlockWorkerClient blockWorkerClient = new DefaultBlockWorkerClient(
         mLocalAlluxioClusterResource.get().getWorkerAddress(), mExecutorService,
         1 /* fake session id */, true, new ClientMetrics())) {
       Assert.assertFalse(blockWorkerClient.isConnected());

tests/src/test/java/alluxio/worker/DataServerIntegrationTest.java

@@ -20,7 +20,7 @@
 import alluxio.client.WriteType;
 import alluxio.client.block.BlockMasterClient;
 import alluxio.client.block.BlockStoreContext;
-import alluxio.client.block.DefaultBlockWorkerClient;
+import alluxio.client.block.BlockWorkerClient;
 import alluxio.client.block.RetryHandlingBlockMasterClient;
 import alluxio.client.file.FileSystem;
 import alluxio.client.file.URIStatus;
@@ -74,7 +74,7 @@ public static Collection<Object[]> data() {
   public LocalAlluxioClusterResource mLocalAlluxioClusterResource;
   private FileSystem mFileSystem = null;
   private BlockMasterClient mBlockMasterClient;
-  private DefaultBlockWorkerClient mBlockWorkerClient;
+  private BlockWorkerClient mBlockWorkerClient;
 
   public DataServerIntegrationTest(String className, String nettyTransferType, String blockReader) {
     mLocalAlluxioClusterResource = new LocalAlluxioClusterResource(WORKER_CAPACITY_BYTES,