### Question:
SepEventExecutor { public void scheduleSepEvent(SepEvent sepEvent) { if (stopped) { throw new IllegalStateException("This executor is stopped"); } int partition = (hashFunction.hashBytes(sepEvent.getRow()).asInt() & Integer.MAX_VALUE) % numThreads; List<SepEvent> eventBuffer = (List<SepEvent>)eventBuffers.get(partition); eventBuffer.add(sepEvent); if (eventBuffer.size() == batchSize) { scheduleEventBatch(partition, Lists.newArrayList(eventBuffer)); eventBuffers.removeAll(partition); } } SepEventExecutor(EventListener eventListener, List<ThreadPoolExecutor> executors, int batchSize, SepMetrics sepMetrics); void scheduleSepEvent(SepEvent sepEvent); List<Future<?>> flush();  }### Answer:
@Test public void testScheduleSepEvent() throws InterruptedException { RecordingEventListener eventListener = new RecordingEventListener(); SepEventExecutor executor = new SepEventExecutor(eventListener, getExecutors(2), 1, sepMetrics); final int NUM_EVENTS = 10; for (int i = 0; i < NUM_EVENTS; i++) { executor.scheduleSepEvent(createSepEvent(i)); } for (int retry = 0; retry < 50; retry++) { if (eventListener.receivedEvents.size() >= NUM_EVENTS) { break; } Thread.sleep(250); } assertEquals(NUM_EVENTS, eventListener.receivedEvents.size()); }