@Override
    public <R> QueryResult<R> query(final Query<R> query, final PositionBound positionBound, final QueryConfig config) {
        return internal.query(query, positionBound, config);
    }
@Test
    public void shouldTrackOpenIteratorsMetric() {
        final MultiVersionedKeyQuery<String, String> query = MultiVersionedKeyQuery.withKey(KEY);
        final PositionBound bound = PositionBound.unbounded();
        final QueryConfig config = new QueryConfig(false);
        when(inner.query(any(), any(), any())).thenReturn(
            QueryResult.forResult(new LogicalSegmentIterator(Collections.emptyListIterator(), RAW_KEY, 0L, 0L, ResultOrder.ANY)));

        final KafkaMetric openIteratorsMetric = getMetric("num-open-iterators");
        assertThat(openIteratorsMetric, not(nullValue()));

        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));

        final QueryResult<VersionedRecordIterator<String>> result = store.query(query, bound, config);

        try (final VersionedRecordIterator<String> iterator = result.getResult()) {
            assertThat((Long) openIteratorsMetric.metricValue(), equalTo(1L));
        }

        assertThat((Long) openIteratorsMetric.metricValue(), equalTo(0L));
    }