@@ -0,0 +1,2681 @@
+package alluxio.worker.block;%0A%0Aimport alluxio.ConfigurationRule;%0Aimport alluxio.PropertyKey;%0Aimport alluxio.underfs.SeekableUnderFileInputStream;%0Aimport alluxio.underfs.UnderFileSystem;%0Aimport alluxio.underfs.options.OpenOptions;%0A%0Aimport org.junit.Assert;%0Aimport org.junit.Before;%0Aimport org.junit.Test;%0Aimport org.mockito.Mockito;%0A%0Aimport java.io.InputStream;%0Aimport java.util.ArrayList;%0Aimport java.util.HashMap;%0Aimport java.util.List;%0A%0Apublic class UfsInputStreamManagerTest %7B%0A  private static final String FILE_NAME = %22/test%22;%0A%0A  private UnderFileSystem mUfs;%0A  private List%3CSeekableUnderFileInputStream%3E mSeekableInStreams;%0A  private UfsInputStreamManager mManager;%0A%0A  public ConfigurationRule mConfigurationRule =%0A      new ConfigurationRule(new HashMap%3CPropertyKey, String%3E() %7B%0A        %7B%0A          put(PropertyKey.WORKER_UFS_INSTREAM_CACHE_EXPIRE_MS, %22200000%22);%0A        %7D%0A      %7D);%0A%0A  @Before%0A  public void before() throws Exception %7B%0A    mSeekableInStreams = new ArrayList%3C%3E();%0A    mUfs = Mockito.mock(UnderFileSystem.class);%0A    Mockito.when(mUfs.isSeekable()).thenReturn(true);%0A    for(int i=0;i%3C3;i++) %7B%0A      SeekableUnderFileInputStream instream=createMockedSeekableStream();%0A      Mockito.when(mUfs.open(Mockito.eq(FILE_NAME), Mockito.any(OpenOptions.class))).thenReturn(instream);%0A      mSeekableInStreams.add(instream);%0A    %7D%0A    mManager = new UfsInputStreamManager();%0A  %7D%0A%0A  private SeekableUnderFileInputStream createMockedSeekableStream() %7B%0A    SeekableUnderFileInputStream instream = Mockito.mock(SeekableUnderFileInputStream.class);%0A    Mockito.when(instream.getFilePath()).thenReturn(FILE_NAME);%0A    Mockito.doCallRealMethod().when(instream).setResourceId(Mockito.any(Long.class));%0A    Mockito.when(instream.getResourceId()).thenCallRealMethod();%0A    return instream;%0A  %7D%0A%0A  @Test%0A  public void testCheckInAndOut() throws Exception %7B%0A    mSeekableInStreams = new ArrayList%3C%3E();%0A    SeekableUnderFileInputStream mockedStrem = createMockedSeekableStream();%0A    Mockito.when(mUfs.open(Mockito.eq(FILE_NAME), Mockito.any(OpenOptions.class))).thenReturn(mockedStrem)%0A        .thenThrow(new IllegalStateException(%22Should only be called once%22));%0A    mSeekableInStreams.add(mockedStrem);%0A%0A    // check out a stream%0A    InputStream instream1 = mManager.checkOut(mUfs, FILE_NAME, 2);%0A    // check in back%0A    mManager.checkIn(instream1);%0A // check out a stream again%0A    InputStream instream2 = mManager.checkOut(mUfs, FILE_NAME, 4);%0A%0A    Assert.assertEquals(mockedStrem, instream1);%0A    Assert.assertEquals(mockedStrem, instream2);%0A    // ensure the second time the checked out instream is the same one but repositioned%0A    Mockito.verify(mockedStrem).seek(4);%0A  %7D%0A%7D%0A

add check-in and then check-out test