@Test(timeout=60000) public void testShortCircuitCacheTemporaryFailure() throws Exception { BlockReaderTestUtil.enableBlockReaderFactoryTracing(); final AtomicBoolean replicaCreationShouldFail = new AtomicBoolean(true); final AtomicBoolean testFailed = new AtomicBoolean(false); DFSInputStream.tcpReadsDisabledForTesting = true; BlockReaderFactory.createShortCircuitReplicaInfoCallback = new ShortCircuitCache.ShortCircuitReplicaCreator() { @Override public ShortCircuitReplicaInfo createShortCircuitReplicaInfo() { if (replicaCreationShouldFail.get()) { Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS); return new ShortCircuitReplicaInfo(); } return null; } }; TemporarySocketDirectory sockDir = new TemporarySocketDirectory(); Configuration conf = createShortCircuitConf( "testShortCircuitCacheTemporaryFailure", sockDir); final MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build(); cluster.waitActive(); final DistributedFileSystem dfs = cluster.getFileSystem(); final String TEST_FILE = "/test_file"; final int TEST_FILE_LEN = 4000; final int NUM_THREADS = 2; final int SEED = 0xFADED; final CountDownLatch gotFailureLatch = new CountDownLatch(NUM_THREADS); final CountDownLatch shouldRetryLatch = new CountDownLatch(1); DFSTestUtil.createFile(dfs, new Path(TEST_FILE), TEST_FILE_LEN, (short)1, SEED); Runnable readerRunnable = new Runnable() { @Override public void run() { try { List<LocatedBlock> locatedBlocks = cluster.getNameNode().getRpcServer().getBlockLocations( TEST_FILE, 0, TEST_FILE_LEN).getLocatedBlocks(); LocatedBlock lblock = locatedBlocks.get(0); BlockReader blockReader = null; try { blockReader = BlockReaderTestUtil. getBlockReader(cluster, lblock, 0, TEST_FILE_LEN); Assert.fail("expected getBlockReader to fail the first time."); } catch (Throwable t) { Assert.assertTrue("expected to see 'TCP reads were disabled " + "for testing' in exception " + t, t.getMessage().contains( "TCP reads were disabled for testing")); } finally { if (blockReader != null) blockReader.close(); } gotFailureLatch.countDown(); shouldRetryLatch.await(); try { blockReader = BlockReaderTestUtil. getBlockReader(cluster, lblock, 0, TEST_FILE_LEN); } catch (Throwable t) { LOG.error("error trying to retrieve a block reader " + "the second time.", t); throw t; } finally { if (blockReader != null) blockReader.close(); } } catch (Throwable t) { LOG.error("getBlockReader failure", t); testFailed.set(true); } } }; Thread threads[] = new Thread[NUM_THREADS]; for (int i = 0; i < NUM_THREADS; i++) { threads[i] = new Thread(readerRunnable); threads[i].start(); } gotFailureLatch.await(); replicaCreationShouldFail.set(false); shouldRetryLatch.countDown(); for (int i = 0; i < NUM_THREADS; i++) { Uninterruptibles.joinUninterruptibly(threads[i]); } cluster.shutdown(); sockDir.close(); Assert.assertFalse(testFailed.get()); }
public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); BlockReaderFactory setFileName(String fileName); BlockReaderFactory setBlock(ExtendedBlock block); BlockReaderFactory setBlockToken(Token<BlockTokenIdentifier> token); BlockReaderFactory setStartOffset(long startOffset); BlockReaderFactory setVerifyChecksum(boolean verifyChecksum); BlockReaderFactory setClientName(String clientName); BlockReaderFactory setDatanodeInfo(DatanodeInfo datanode); BlockReaderFactory setAllowShortCircuitLocalReads(
      boolean allowShortCircuitLocalReads); BlockReaderFactory setClientCacheContext(
      ClientContext clientContext); BlockReaderFactory setLength(long length); BlockReaderFactory setCachingStrategy(
      CachingStrategy cachingStrategy); BlockReaderFactory setInetSocketAddress(
      InetSocketAddress inetSocketAddress); BlockReaderFactory setUserGroupInformation(
      UserGroupInformation userGroupInformation); BlockReaderFactory setRemotePeerFactory(
      RemotePeerFactory remotePeerFactory); BlockReaderFactory setConfiguration(
      Configuration configuration); BlockReaderFactory setTimestamp(
      long timestamp); BlockReaderFactory setBUserTimestamp(boolean bUserTimestamp); BlockReader build(); @Override ShortCircuitReplicaInfo createShortCircuitReplicaInfo(); @Override String toString(); static String getFileName(final InetSocketAddress s,
      final String poolId, final long blockId); }
BlockReaderFactory implements ShortCircuitReplicaCreator { public BlockReader build() throws IOException { BlockReader reader = null; Preconditions.checkNotNull(configuration); if (conf.shortCircuitLocalReads && allowShortCircuitLocalReads) { if (clientContext.getUseLegacyBlockReaderLocal()) { reader = getLegacyBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new legacy block reader local."); } return reader; } } else { reader = getBlockReaderLocal(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new block reader local."); } return reader; } } } if (conf.domainSocketDataTraffic) { reader = getRemoteBlockReaderFromDomain(); if (reader != null) { if (LOG.isTraceEnabled()) { LOG.trace(this + ": returning new remote block reader using " + "UNIX domain socket on " + pathInfo.getPath()); } return reader; } } Preconditions.checkState(!DFSInputStream.tcpReadsDisabledForTesting, "TCP reads were disabled for testing, but we failed to " + "do a non-TCP read."); return getRemoteBlockReaderFromTcp(); } BlockReaderFactory(DFSClient.Conf conf); BlockReaderFactory setFileName(String fileName); BlockReaderFactory setBlock(ExtendedBlock block); BlockReaderFactory setBlockToken(Token<BlockTokenIdentifier> token); BlockReaderFactory setStartOffset(long startOffset); BlockReaderFactory setVerifyChecksum(boolean verifyChecksum); BlockReaderFactory setClientName(String clientName); BlockReaderFactory setDatanodeInfo(DatanodeInfo datanode); BlockReaderFactory setAllowShortCircuitLocalReads(
      boolean allowShortCircuitLocalReads); BlockReaderFactory setClientCacheContext(
      ClientContext clientContext); BlockReaderFactory setLength(long length); BlockReaderFactory setCachingStrategy(
      CachingStrategy cachingStrategy); BlockReaderFactory setInetSocketAddress(
      InetSocketAddress inetSocketAddress); BlockReaderFactory setUserGroupInformation(
      UserGroupInformation userGroupInformation); BlockReaderFactory setRemotePeerFactory(
      RemotePeerFactory remotePeerFactory); BlockReaderFactory setConfiguration(
      Configuration configuration); BlockReaderFactory setTimestamp(
      long timestamp); BlockReaderFactory setBUserTimestamp(boolean bUserTimestamp); BlockReader build(); @Override ShortCircuitReplicaInfo createShortCircuitReplicaInfo(); @Override String toString(); static String getFileName(final InetSocketAddress s,
      final String poolId, final long blockId);  }