@Override
  public void close() {
    loggerFactory.getLoggersWithPrefix(queryId.toString()).forEach(ProcessingLogger::close);
    sharedKafkaStreamsRuntime.stop(queryId, true);
    scalablePushRegistry.ifPresent(ScalablePushRegistry::close);
    listener.onClose(this);
  }
@Test
    public void shouldCloseProcessingLoggers() {
        // Given:
        final ProcessingLogger processingLogger1 = mock(ProcessingLogger.class);
        final ProcessingLogger processingLogger2 = mock(ProcessingLogger.class);
        when(loggerFactory.getLoggersWithPrefix(QUERY_ID.toString())).thenReturn(Arrays.asList(processingLogger1, processingLogger2));

        // When:
        query.close();

        // Then:
        verify(processingLogger1).close();
        verify(processingLogger2).close();
    }