public static <V, K> void produceSynchronously(final Properties producerConfig,
                                                   final boolean eos,
                                                   final String topic,
                                                   final Optional<Integer> partition,
                                                   final List<KeyValueTimestamp<K, V>> toProduce) {
        try (final Producer<K, V> producer = new KafkaProducer<>(producerConfig)) {
            if (eos) {
                producer.initTransactions();
                producer.beginTransaction();
            }
            final LinkedList<Future<RecordMetadata>> futures = new LinkedList<>();
            for (final KeyValueTimestamp<K, V> record : toProduce) {
                final Future<RecordMetadata> f = producer.send(
                    new ProducerRecord<>(
                        topic,
                        partition.orElse(null),
                        record.timestamp(),
                        record.key(),
                        record.value(),
                        null
                    )
                );
                futures.add(f);
            }

            if (eos) {
                producer.commitTransaction();
            } else {
                producer.flush();
            }

            for (final Future<RecordMetadata> future : futures) {
                try {
                    future.get();
                } catch (final InterruptedException | ExecutionException e) {
                    throw new RuntimeException(e);
                }
            }
        }
    }
@param topic               Kafka topic to write the data records to
@param records             Data records to write to Kafka
@param producerConfig      Kafka producer configuration
@param headers             {@link Headers} of the data records
@param timestamp           Timestamp of the record
@param enableTransactions  Send messages in a transaction
@param <K>                 Key type of the data records
@param <V>                 Value type of the data records
streams/src/test/java/org/apache/kafka/streams/integration/utils/IntegrationTestUtils.java