public static <K, V> void produceKeyValuesSynchronously(final String topic,
                                                            final Collection<KeyValue<K, V>> records,
                                                            final Properties producerConfig,
                                                            final Headers headers,
                                                            final Time time,
                                                            final boolean enableTransactions) {
        try (final Producer<K, V> producer = new KafkaProducer<>(producerConfig)) {
            if (enableTransactions) {
                producer.initTransactions();
                producer.beginTransaction();
            }
            for (final KeyValue<K, V> record : records) {
                producer.send(new ProducerRecord<>(topic, null, time.milliseconds(), record.key, record.value, headers));
                time.sleep(1L);
            }
            if (enableTransactions) {
                producer.commitTransaction();
            } else {
                producer.flush();
            }
        }
    }
@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 time                Timestamp provider
@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