OfframpServiceV2 implements OfframpService { @Override public void run() { try { sendEvent(new Connection(podName)); long timeout = 0L; while (!shuttingDown) { Event event = events.poll(timeout, MILLISECONDS); timeout = 0L; if (event != null) { handleIncomingEvent(event); } else if (requested > 0L) { if (buffer.isEmpty()) { replenishBuffer(); } Record record = buffer.poll(); if (record != null) { sendMessage(record); } else { timeout = 1L; } } else { timeout = 1L; } } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } catch (Exception e) { log.error("An error occurred while serving", e); throw new RuntimeException(e); } finally { semaphore.release(); } }  @Override void run(); @Override void close(); @Override void onEvent(String event);  }
@Test public void run() throws Exception { Request request = new Request(1L); JsonNode value = mapper.createObjectNode(); Payload<JsonNode> payload = new Payload<JsonNode>((byte) 0, 2, value); Record record = new Record(0, 1L, 3L, payload); doNothing().when(underTest).sendEvent(any()); doNothing().when(underTest).sendRebalance(any()); doAnswer(i -> { underTest.getBuffer().offer(record); return null; }).when(underTest).replenishBuffer(); try { Mono.fromRunnable(underTest).subscribeOn(single()).subscribe(); underTest.getEvents().offer(request); Awaitility.await().atMost(500, MILLISECONDS).pollInterval(10, MILLISECONDS).until(() -> { verify(underTest).handleIncomingEvent(request); }); Awaitility.await().atMost(500, MILLISECONDS).pollInterval(10, MILLISECONDS).until(() -> { verify(underTest).sendMessage(record); }); } finally { underTest.close(); } verify(underTest).sendEvent(new Connection(podName)); ArgumentCaptor<RebalanceListener> captor = ArgumentCaptor.forClass(RebalanceListener.class); verify(consumer).init(eq(1L), captor.capture()); Set<Integer> assignment = singleton(0); captor.getValue().onRebalance(assignment); verify(underTest).sendRebalance(assignment); }