class StreamsProducer {

    void commitTransaction(final Map<TopicPartition, OffsetAndMetadata> offsets,
                           final ConsumerGroupMetadata consumerGroupMetadata) throws ProducerFencedException {
        if (!eosEnabled()) {
            throw new IllegalStateException(formatException("Exactly-once is not enabled"));
        }
        maybeBeginTransaction();
        try {
            producer.sendOffsetsToTransaction(offsets, consumerGroupMetadata);
            producer.commitTransaction();
            transactionInFlight = false;
        } catch (final ProducerFencedException | CommitFailedException error) {
            throw new TaskMigratedException(
                formatException("Producer got fenced trying to commit a transaction"),
                error
            );
        } catch (final TimeoutException error) {
            // TODO KIP-447: we can consider treating it as non-fatal and retry on the thread level
            throw new StreamsException(formatException("Timed out trying to commit a transaction"), error);
        } catch (final KafkaException error) {
            throw new StreamsException(
                formatException("Error encountered trying to commit a transaction"),
                error
            );
        }
    }

    public  StreamsProducer(final StreamsConfig config,
                           final String threadId,
                           final KafkaClientSupplier clientSupplier,
                           final TaskId taskId,
                           final UUID processId,
                           final LogContext logContext);

}