void commitOffsetsOrTransaction(final Map<Task, Map<TopicPartition, OffsetAndMetadata>> offsetsPerTask) {
        log.debug("Committing task offsets {}", offsetsPerTask.entrySet().stream().collect(Collectors.toMap(t -> t.getKey().id(), Entry::getValue))); // avoid logging actual Task objects

        final Set<TaskId> corruptedTasks = new HashSet<>();

        if (executionMetadata.processingMode() == EXACTLY_ONCE_ALPHA) {
            for (final Task task : taskManager.activeRunningTaskIterable()) {
                final Map<TopicPartition, OffsetAndMetadata> taskOffsetsToCommit = offsetsPerTask.getOrDefault(task, emptyMap());
                if (!taskOffsetsToCommit.isEmpty() || taskManager.streamsProducerForTask(task.id()).transactionInFlight()) {
                    try {
                        taskManager.streamsProducerForTask(task.id())
                            .commitTransaction(taskOffsetsToCommit, taskManager.consumerGroupMetadata());
                        updateTaskCommitMetadata(taskOffsetsToCommit);
                    } catch (final TimeoutException timeoutException) {
                        log.error(
                            String.format("Committing task %s failed.", task.id()),
                            timeoutException
                        );
                        corruptedTasks.add(task.id());
                    }
                }
            }
        } else if (executionMetadata.processingMode() == EXACTLY_ONCE_V2) {
            if (!offsetsPerTask.isEmpty() || taskManager.threadProducer().transactionInFlight()) {
                final Map<TopicPartition, OffsetAndMetadata> allOffsets = offsetsPerTask.values().stream()
                    .flatMap(e -> e.entrySet().stream()).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

                try {
                    taskManager.threadProducer().commitTransaction(allOffsets, taskManager.consumerGroupMetadata());
                    updateTaskCommitMetadata(allOffsets);
                } catch (final TimeoutException timeoutException) {
                    log.error(
                        String.format("Committing task(s) %s failed.",
                                      offsetsPerTask
                                          .keySet()
                                          .stream()
                                          .map(t -> t.id().toString())
                                          .collect(Collectors.joining(", "))),
                        timeoutException
                    );
                    offsetsPerTask
                        .keySet()
                        .forEach(task -> corruptedTasks.add(task.id()));
                }
            }
        } else {
            // processingMode == ALOS
            if (!offsetsPerTask.isEmpty()) {
                final Map<TopicPartition, OffsetAndMetadata> allOffsets = offsetsPerTask.values().stream()
                    .flatMap(e -> e.entrySet().stream()).collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

                try {
                    taskManager.consumerCommitSync(allOffsets);
                    updateTaskCommitMetadata(allOffsets);
                } catch (final CommitFailedException error) {
                    throw new TaskMigratedException("Consumer committing offsets failed, " +
                                                        "indicating the corresponding thread is no longer part of the group", error);
                } catch (final TimeoutException timeoutException) {
                    log.error(
                        String.format("Committing task(s) %s failed.",
                                      offsetsPerTask
                                          .keySet()
                                          .stream()
                                          .map(t -> t.id().toString())
                                          .collect(Collectors.joining(", "))),
                        timeoutException
                    );
                    throw timeoutException;
                } catch (final KafkaException error) {
                    throw new StreamsException("Error encountered committing offsets via consumer", error);
                }
            }
        }
        if (!corruptedTasks.isEmpty()) {
            throw new TaskCorruptedException(corruptedTasks);
        }
    }
@Test
    public void testCommitWithOpenTransactionButNoOffsetsEOSV2() {
        final Tasks tasks = mock(Tasks.class);
        final TaskManager taskManager = mock(TaskManager.class);
        final ConsumerGroupMetadata groupMetadata = mock(ConsumerGroupMetadata.class);
        when(taskManager.consumerGroupMetadata()).thenReturn(groupMetadata);

        final TaskExecutionMetadata metadata = mock(TaskExecutionMetadata.class);
        final StreamsProducer producer = mock(StreamsProducer.class);
        when(metadata.processingMode()).thenReturn(EXACTLY_ONCE_V2);
        when(taskManager.threadProducer()).thenReturn(producer);
        when(producer.transactionInFlight()).thenReturn(true);

        final TaskExecutor taskExecutor = new TaskExecutor(tasks, taskManager, metadata, new LogContext());
        taskExecutor.commitOffsetsOrTransaction(Collections.emptyMap());

        verify(producer).commitTransaction(Collections.emptyMap(), groupMetadata);
    }