public void produce(final String kafka, final Duration timeToSpend) throws InterruptedException {
            final Properties producerProps = new Properties();
            final String id = "RelationalSmokeTestProducer" + UUID.randomUUID();
            producerProps.put(ProducerConfig.CLIENT_ID_CONFIG, id);
            producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
            producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, IntegerSerializer.class);
            producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
            producerProps.put(ProducerConfig.ACKS_CONFIG, "-1");

            final Article.ArticleSerializer articleSerializer = new Article.ArticleSerializer();
            final Comment.CommentSerializer commentSerializer = new Comment.CommentSerializer();

            final long pauseTime = timeToSpend.toMillis() / (articles.length + comments.length);

            try (final KafkaProducer<Integer, byte[]> producer = new KafkaProducer<>(producerProps)) {
                int a = 0;
                int c = 0;
                while (a < articles.length || c < comments.length) {
                    final ProducerRecord<Integer, byte[]> producerRecord;
                    if (a < articles.length && c >= comments.length ||
                        a < articles.length && articles[a].getTimestamp() <= comments[c].timestamp) {
                        producerRecord = new ProducerRecord<>(
                            ARTICLE_SOURCE,
                            null,
                            articles[a].getTimestamp(),
                            articles[a].getKey(),
                            articleSerializer.serialize("", articles[a])
                        );
                        a++;
                    } else {
                        producerRecord = new ProducerRecord<>(
                            COMMENT_SOURCE,
                            null,
                            comments[c].getTimestamp(),
                            comments[c].getKey(),
                            commentSerializer.serialize("", comments[c])
                        );
                        c++;
                    }
                    producer.send(producerRecord);
                    producer.flush();
                    LOG.info("sent {} {}", producerRecord.topic(), producerRecord.key());
                    Thread.sleep(pauseTime);
                }
            }
        }
Generates a keySpace number of unique keys normally distributed,
with the mean at 0 and stdDev 1/3 of the way through the keySpace
any sample more than 3 standard deviations from the mean are assigned to the 0 key.
<p>
This is designed to roughly balance the key properties of two dominant real-world
data distribution: Zipfian and Normal, while also being efficient to generate.
streams/src/test/java/org/apache/kafka/streams/tests/RelationalSmokeTest.java