@Override
    public void unsubscribe() {
        acquireAndEnsureOpen();
        try {
            fetchBuffer.retainAll(Collections.emptySet());
            Timer timer = time.timer(Long.MAX_VALUE);
            UnsubscribeEvent unsubscribeEvent = new UnsubscribeEvent(calculateDeadlineMs(timer));
            applicationEventHandler.add(unsubscribeEvent);
            log.info("Unsubscribing all topics or patterns and assigned partitions {}",
                    subscriptions.assignedPartitions());

            try {
                processBackgroundEvents(unsubscribeEvent.future(), timer);
                log.info("Unsubscribed all topics or patterns and assigned partitions");
            } catch (TimeoutException e) {
                log.error("Failed while waiting for the unsubscribe event to complete");
            }
            resetGroupMetadata();
        } catch (Exception e) {
            log.error("Unsubscribe failed", e);
            throw e;
        } finally {
            release();
        }
    }
@Test
    public void testUnsubscribeWithoutGroupId() {
        consumer = newConsumerWithoutGroupId();

        completeUnsubscribeApplicationEventSuccessfully();
        consumer.unsubscribe();
        verify(applicationEventHandler).add(ArgumentMatchers.isA(UnsubscribeEvent.class));
    }