BlockListAsLongs implements Iterable<BlockReportReplica> { @VisibleForTesting public static BlockListAsLongs encode( final Collection<? extends Replica> replicas) { BlockListAsLongs.Builder builder = builder(IPC_MAXIMUM_DATA_LENGTH_DEFAULT); for (Replica replica : replicas) { builder.add(replica); } return builder.build(); }  static BlockListAsLongs decodeBuffer(final int numBlocks,
      final ByteString blocksBuf, final int maxDataLength); @VisibleForTesting static BlockListAsLongs decodeBuffers(final int numBlocks,
      final List<ByteString> blocksBufs); static BlockListAsLongs decodeBuffers(final int numBlocks,
      final List<ByteString> blocksBufs, final int maxDataLength); static BlockListAsLongs decodeLongs(List<Long> blocksList); static BlockListAsLongs decodeLongs(List<Long> blocksList,
      int maxDataLength); @VisibleForTesting static BlockListAsLongs encode(
      final Collection<? extends Replica> replicas); static BlockListAsLongs readFrom(InputStream is, int maxDataLength); void writeTo(OutputStream os); @VisibleForTesting static Builder builder(); static Builder builder(int maxDataLength); abstract int getNumberOfBlocks(); abstract ByteString getBlocksBuffer(); List<ByteString> getBlocksBuffers(); abstract long[] getBlockListAsLongs(); abstract Iterator<BlockReportReplica> iterator(); static BlockListAsLongs EMPTY; }
@Test public void testDatanodeDetect() throws ServiceException, IOException { final AtomicReference<BlockReportRequestProto> request = new AtomicReference<>(); DatanodeProtocolPB mockProxy = mock(DatanodeProtocolPB.class); doAnswer(new Answer<BlockReportResponseProto>() { public BlockReportResponseProto answer(InvocationOnMock invocation) { Object[] args = invocation.getArguments(); request.set((BlockReportRequestProto) args[1]); return BlockReportResponseProto.newBuilder().build(); } }).when(mockProxy).blockReport(any(), any(BlockReportRequestProto.class)); @SuppressWarnings("resource") DatanodeProtocolClientSideTranslatorPB nn = new DatanodeProtocolClientSideTranslatorPB(mockProxy); DatanodeRegistration reg = DFSTestUtil.getLocalDatanodeRegistration(); NamespaceInfo nsInfo = new NamespaceInfo(1, "cluster", "bp", 1); reg.setNamespaceInfo(nsInfo); Replica r = new FinalizedReplica(new Block(1, 2, 3), null, null); BlockListAsLongs bbl = BlockListAsLongs.encode(Collections.singleton(r)); DatanodeStorage storage = new DatanodeStorage("s1"); StorageBlockReport[] sbr = { new StorageBlockReport(storage, bbl) }; request.set(null); nsInfo.setCapabilities(Capability.STORAGE_BLOCK_REPORT_BUFFERS.getMask()); nn.blockReport(reg, "pool", sbr, new BlockReportContext(1, 0, System.nanoTime(), 0L, true)); BlockReportRequestProto proto = request.get(); assertNotNull(proto); assertTrue(proto.getReports(0).getBlocksList().isEmpty()); assertFalse(proto.getReports(0).getBlocksBuffersList().isEmpty()); request.set(null); nsInfo.setCapabilities(Capability.UNKNOWN.getMask()); BlockListAsLongs blockList = getBlockList(r); StorageBlockReport[] obp = new StorageBlockReport[] { new StorageBlockReport(new DatanodeStorage("s1"), blockList) }; nn.blockReport(reg, "pool", obp, new BlockReportContext(1, 0, System.nanoTime(), 0L, true)); proto = request.get(); assertNotNull(proto); assertFalse(proto.getReports(0).getBlocksList().isEmpty()); assertTrue(proto.getReports(0).getBlocksBuffersList().isEmpty()); }