ShuffleHandler extends AuxiliaryService { protected Shuffle getShuffle(Configuration conf) { return new Shuffle(conf); } ShuffleHandler(MetricsSystem ms);  ShuffleHandler(); static ByteBuffer serializeMetaData(int port); static int deserializeMetaData(ByteBuffer meta); static ByteBuffer serializeServiceData(Token<JobTokenIdentifier> jobToken); @Override void initializeApplication(ApplicationInitializationContext context); @Override void stopApplication(ApplicationTerminationContext context); @Override synchronized ByteBuffer getMetaData(); static final String SHUFFLE_MANAGE_OS_CACHE; static final boolean DEFAULT_SHUFFLE_MANAGE_OS_CACHE; static final String SHUFFLE_READAHEAD_BYTES; static final int DEFAULT_SHUFFLE_READAHEAD_BYTES; static final HttpResponseStatus TOO_MANY_REQ_STATUS; static final long FETCH_RETRY_DELAY; static final String RETRY_AFTER_HEADER; static final String MAPREDUCE_SHUFFLE_SERVICEID; static final String SHUFFLE_PORT_CONFIG_KEY; static final int DEFAULT_SHUFFLE_PORT; static final String SHUFFLE_LISTEN_QUEUE_SIZE; static final int DEFAULT_SHUFFLE_LISTEN_QUEUE_SIZE; static final String SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED; static final boolean DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_ENABLED; static final String SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT; static final int DEFAULT_SHUFFLE_CONNECTION_KEEP_ALIVE_TIME_OUT; static final String SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE; static final int DEFAULT_SHUFFLE_MAPOUTPUT_META_INFO_CACHE_SIZE; static final String CONNECTION_CLOSE; static final String SUFFLE_SSL_FILE_BUFFER_SIZE_KEY; static final int DEFAULT_SUFFLE_SSL_FILE_BUFFER_SIZE; static final String MAX_SHUFFLE_CONNECTIONS; static final int DEFAULT_MAX_SHUFFLE_CONNECTIONS; static final String MAX_SHUFFLE_THREADS; static final int DEFAULT_MAX_SHUFFLE_THREADS; static final String SHUFFLE_BUFFER_SIZE; static final int DEFAULT_SHUFFLE_BUFFER_SIZE; static final String SHUFFLE_TRANSFERTO_ALLOWED; static final boolean DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED; static final boolean WINDOWS_DEFAULT_SHUFFLE_TRANSFERTO_ALLOWED; static final String SHUFFLE_MAX_SESSION_OPEN_FILES; static final int DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES; }
@Test(timeout = 4000) public void testSendMapCount() throws Exception { final List<ShuffleHandler.ReduceMapFileCount> listenerList = new ArrayList<ShuffleHandler.ReduceMapFileCount>(); final ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class); final MessageEvent mockEvt = mock(MessageEvent.class); final Channel mockCh = mock(AbstractChannel.class); final ChannelPipeline mockPipeline = mock(ChannelPipeline.class); final HttpRequest mockHttpRequest = createMockHttpRequest(); final ChannelFuture mockFuture = createMockChannelFuture(mockCh, listenerList); final ShuffleHandler.TimeoutHandler timerHandler = new ShuffleHandler.TimeoutHandler(); Mockito.doReturn(mockCh).when(mockCtx).getChannel(); when(mockCh.getPipeline()).thenReturn(mockPipeline); when(mockPipeline.get( Mockito.any(String.class))).thenReturn(timerHandler); when(mockCtx.getChannel()).thenReturn(mockCh); Mockito.doReturn(mockFuture).when(mockCh).write(Mockito.any(Object.class)); when(mockCh.write(Object.class)).thenReturn(mockFuture); Mockito.doReturn(mockCh).when(mockEvt).getChannel(); when(mockEvt.getChannel()).thenReturn(mockCh); Mockito.doReturn(mockHttpRequest).when(mockEvt).getMessage(); final ShuffleHandler sh = new MockShuffleHandler(); Configuration conf = new Configuration(); sh.init(conf); sh.start(); int maxOpenFiles =conf.getInt(ShuffleHandler.SHUFFLE_MAX_SESSION_OPEN_FILES, ShuffleHandler.DEFAULT_SHUFFLE_MAX_SESSION_OPEN_FILES); sh.getShuffle(conf).messageReceived(mockCtx, mockEvt); assertTrue("Number of Open files should not exceed the configured " + "value!-Not Expected", listenerList.size() <= maxOpenFiles); while(!listenerList.isEmpty()) { listenerList.remove(0).operationComplete(mockFuture); assertTrue("Number of Open files should not exceed the configured " + "value!-Not Expected", listenerList.size() <= maxOpenFiles); } sh.close(); }