public static boolean verifySync(final String broker, final Instant deadline) throws InterruptedException {
            final Deserializer<Integer> keyDeserializer = intSerde.deserializer();

            final Deserializer<Article> articleDeserializer = new Article.ArticleDeserializer();

            final Deserializer<AugmentedArticle> augmentedArticleDeserializer =
                new AugmentedArticle.AugmentedArticleDeserializer();

            final Deserializer<Comment> commentDeserializer = new Comment.CommentDeserializer();

            final Deserializer<AugmentedComment> augmentedCommentDeserializer =
                new AugmentedComment.AugmentedCommentDeserializer();


            final Properties consumerProperties = new Properties();
            final String id = "RelationalSmokeTestConsumer" + UUID.randomUUID();
            consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG, id);
            consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, id);
            consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, broker);
            consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, IntegerDeserializer.class);
            consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
            consumerProperties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
            consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

            try (final KafkaConsumer<Integer, byte[]> consumer = new KafkaConsumer<>(consumerProperties)) {
                final List<PartitionInfo> articlePartitions = consumer.partitionsFor(ARTICLE_SOURCE);
                final List<PartitionInfo> augmentedArticlePartitions = consumer.partitionsFor(ARTICLE_RESULT_SINK);
                final List<PartitionInfo> commentPartitions = consumer.partitionsFor(COMMENT_SOURCE);
                final List<PartitionInfo> augmentedCommentPartitions = consumer.partitionsFor(COMMENT_RESULT_SINK);
                final List<TopicPartition> assignment =
                    Stream.concat(
                        Stream.concat(
                            articlePartitions.stream().map(p -> new TopicPartition(p.topic(), p.partition())),
                            augmentedArticlePartitions.stream().map(p -> new TopicPartition(p.topic(), p.partition()))
                        ), 
                        Stream.concat(
                            commentPartitions.stream().map(p -> new TopicPartition(p.topic(), p.partition())),
                            augmentedCommentPartitions.stream().map(p -> new TopicPartition(p.topic(), p.partition()))
                        )
                    ).collect(toList());
                consumer.assign(assignment);
                consumer.seekToBeginning(assignment);

                final Map<Integer, Article> consumedArticles = new TreeMap<>();
                final Map<Integer, AugmentedArticle> consumedAugmentedArticles = new TreeMap<>();
                final Map<Integer, Comment> consumedComments = new TreeMap<>();
                final Map<Integer, AugmentedComment> consumedAugmentedComments = new TreeMap<>();

                boolean printedConsumedArticle = false;
                boolean printedConsumedAugmentedArticle = false;
                boolean printedConsumedComment = false;
                boolean printedConsumedAugmentedComment = false;
                boolean passed = false;

                while (!passed && Instant.now().isBefore(deadline)) {
                    boolean lastPollWasEmpty = false;
                    while (!lastPollWasEmpty) {
                        final ConsumerRecords<Integer, byte[]> poll = consumer.poll(Duration.ofSeconds(1));
                        lastPollWasEmpty = poll.isEmpty();
                        for (final ConsumerRecord<Integer, byte[]> record : poll) {
                            final Integer key = record.key();
                            switch (record.topic()) {
                                case ARTICLE_SOURCE: {
                                    final Article article = articleDeserializer.deserialize("", record.value());
                                    if (consumedArticles.containsKey(key)) {
                                        LOG.warn("Duplicate article: {} and {}", consumedArticles.get(key), article);
                                    }
                                    consumedArticles.put(key, article);
                                    break;
                                }
                                case COMMENT_SOURCE: {
                                    final Comment comment = commentDeserializer.deserialize("", record.value());
                                    if (consumedComments.containsKey(key)) {
                                        LOG.warn("Duplicate comment: {} and {}", consumedComments.get(key), comment);
                                    }
                                    consumedComments.put(key, comment);
                                    break;
                                }
                                case ARTICLE_RESULT_SINK: {
                                    final AugmentedArticle article =
                                        augmentedArticleDeserializer.deserialize("", record.value());
                                    consumedAugmentedArticles.put(key, article);
                                    break;
                                }
                                case COMMENT_RESULT_SINK: {
                                    final AugmentedComment comment =
                                        augmentedCommentDeserializer.deserialize("", record.value());
                                    consumedAugmentedComments.put(key, comment);
                                    break;
                                }
                                default:
                                    throw new IllegalArgumentException(record.toString());
                            }
                        }
                        consumer.commitSync();
                    }

                    if (!printedConsumedArticle && !consumedArticles.isEmpty()) {
                        LOG.info("Consumed first Article");
                        printedConsumedArticle = true;
                    }
                    if (!printedConsumedComment && !consumedComments.isEmpty()) {
                        LOG.info("Consumed first Comment");
                        printedConsumedComment = true;
                    }
                    if (!printedConsumedAugmentedArticle && !consumedAugmentedArticles.isEmpty()) {
                        LOG.info("Consumed first AugmentedArticle");
                        printedConsumedAugmentedArticle = true;
                    }
                    if (!printedConsumedAugmentedComment && !consumedAugmentedComments.isEmpty()) {
                        LOG.info("Consumed first AugmentedComment");
                        printedConsumedAugmentedComment = true;
                    }

                    passed = verifySync(
                        false,
                        consumedArticles,
                        consumedComments,
                        consumedAugmentedArticles,
                        consumedAugmentedComments
                    );
                    if (!passed) {
                        LOG.info("Verification has not passed yet. ");
                        Thread.sleep(500);
                    }
                }
                return verifySync(
                    true,
                    consumedArticles,
                    consumedComments,
                    consumedAugmentedArticles,
                    consumedAugmentedComments
                );
            }
        }
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