CoreIngestionGroup implements IngestionGroup { protected AsyncFuture<Ingestion> syncWrite(final Request request) { final Span span = tracer.spanBuilder("CoreIngestionGroup.syncWrite").startSpan(); if (!filter.get().apply(request.getSeries())) { reporter.reportDroppedByFilter(); span.setStatus(Status.FAILED_PRECONDITION.withDescription("Dropped by filter")); span.end(); return async.resolved(Ingestion.of(ImmutableList.of())); } try { span.addAnnotation("Acquiring write lock"); writePermits.acquire(); } catch (final InterruptedException e) { String error = "Failed to acquire semaphore for bounded request"; span.setStatus(Status.INTERNAL.withDescription(error)); span.end(); return async.failed(new Exception(error, e)); } span.addAnnotation("Acquired write lock"); reporter.incrementConcurrentWrites(); try (Scope ws = tracer.withSpan(span)) { return doWrite(request).onFinished(() -> { writePermits.release(); reporter.decrementConcurrentWrites(); span.end(); }); } } CoreIngestionGroup(
        final AsyncFramework async,
        final Supplier<Filter> filter,
        final Semaphore writePermits,
        final IngestionManagerReporter reporter,
        final LongAdder ingested,
        final Optional<MetricBackend> metric,
        final Optional<MetadataBackend> metadata,
        final Optional<SuggestBackend> suggest
    ); @Override Groups groups(); @Override AsyncFuture<Ingestion> write(final Request request); @Override boolean isEmpty();  }
@Test public void testSyncWrite() throws Exception { final CoreIngestionGroup group = setupIngestionGroup(empty(), empty(), empty()); doReturn(filter).when(filterSupplier).get(); doReturn(true).when(filter).apply(series); doNothing().when(writePermits).acquire(); doNothing().when(writePermits).release(); doReturn(expected).when(group).doWrite(request); assertEquals(expected, group.syncWrite(request)); verify(async, never()).resolved(any(Ingestion.class)); verify(async, never()).failed(any(Throwable.class)); verify(writePermits).acquire(); verify(writePermits).release(); verify(reporter).incrementConcurrentWrites(); verify(reporter).decrementConcurrentWrites(); verify(group).doWrite(request); verify(expected).onFinished(any(FutureFinished.class)); }