StreamsProducer streamsProducerForTask(final TaskId taskId) {
        return activeTaskCreator.streamsProducerForTask(taskId);
    }
@Test
    public void shouldCommitViaProducerIfEosAlphaEnabled() {
        final StreamsProducer producer = mock(StreamsProducer.class);
        when(activeTaskCreator.streamsProducerForTask(any(TaskId.class)))
            .thenReturn(producer);

        final Map<TopicPartition, OffsetAndMetadata> offsetsT01 = singletonMap(t1p1, new OffsetAndMetadata(0L, null));
        final Map<TopicPartition, OffsetAndMetadata> offsetsT02 = singletonMap(t1p2, new OffsetAndMetadata(1L, null));

        shouldCommitViaProducerIfEosEnabled(ProcessingMode.EXACTLY_ONCE_ALPHA, offsetsT01, offsetsT02);

        verify(producer).commitTransaction(offsetsT01, new ConsumerGroupMetadata("appId"));
        verify(producer).commitTransaction(offsetsT02, new ConsumerGroupMetadata("appId"));
        verifyNoMoreInteractions(producer);
    }