@Override
    public void close() {
        close(Duration.ofMillis(DEFAULT_CLOSE_TIMEOUT_MS));
    }
@Test
    public void testUnsubscribeOnClose() {
        SubscriptionState subscriptions = mock(SubscriptionState.class);
        consumer = spy(newConsumer(
            mock(FetchBuffer.class),
            mock(ConsumerInterceptors.class),
            mock(ConsumerRebalanceListenerInvoker.class),
            subscriptions,
            "group-id",
            "client-id"));
        completeUnsubscribeApplicationEventSuccessfully();
        consumer.close(Duration.ZERO);
        verifyUnsubscribeEvent(subscriptions);
    }