protected void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,
                                     final ConsumerGroupMetadata consumerGroupMetadata) {
        if (!eosEnabled()) {
            throw new IllegalStateException(formatException("Exactly-once is not enabled"));
        }
        maybeBeginTransaction();
        try {
            // EOS-v2 assumes brokers are on version 2.5+ and thus can understand the full set of consumer group metadata
            // Thus if we are using EOS-v1 and can't make this assumption, we must downgrade the request to include only the group id metadata
            final ConsumerGroupMetadata maybeDowngradedGroupMetadata = processingMode == EXACTLY_ONCE_V2 ? consumerGroupMetadata : new ConsumerGroupMetadata(consumerGroupMetadata.groupId());
            producer.sendOffsetsToTransaction(offsets, maybeDowngradedGroupMetadata);
            producer.commitTransaction();
            transactionInFlight = false;
        } catch (final ProducerFencedException | InvalidProducerEpochException | CommitFailedException | InvalidPidMappingException error) {
            throw new TaskMigratedException(
                formatException("Producer got fenced trying to commit a transaction"),
                error
            );
        } catch (final TimeoutException timeoutException) {
            // re-throw to trigger `task.timeout.ms`
            throw timeoutException;
        } catch (final KafkaException error) {
            throw new StreamsException(
                formatException("Error encountered trying to commit a transaction"),
                error
            );
        }
    }
@throws IllegalStateException if EOS is disabled
@throws TaskMigratedException
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamsProducer.java