public void testConsume(final long prodTimeMs) throws Throwable {
        Properties consumerProps = new Properties();
        consumerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, testConfig.bootstrapServer);
        consumerProps.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 512);
        ClientCompatibilityTestDeserializer deserializer =
            new ClientCompatibilityTestDeserializer(testConfig.expectClusterId);
        try (final KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps, deserializer, deserializer)) {
            final List<PartitionInfo> partitionInfos = consumer.partitionsFor(testConfig.topic);
            if (partitionInfos.isEmpty())
                throw new RuntimeException("Expected at least one partition for topic " + testConfig.topic);
            final Map<TopicPartition, Long> timestampsToSearch = new HashMap<>();
            final LinkedList<TopicPartition> topicPartitions = new LinkedList<>();
            for (PartitionInfo partitionInfo : partitionInfos) {
                TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
                timestampsToSearch.put(topicPartition, prodTimeMs);
                topicPartitions.add(topicPartition);
            }
            final OffsetsForTime offsetsForTime = new OffsetsForTime();
            tryFeature("offsetsForTimes", testConfig.offsetsForTimesSupported,
                () -> offsetsForTime.result = consumer.offsetsForTimes(timestampsToSearch),
                () -> log.info("offsetsForTime = {}", offsetsForTime.result));
            // Whether or not offsetsForTimes works, beginningOffsets and endOffsets
            // should work.
            consumer.beginningOffsets(timestampsToSearch.keySet());
            consumer.endOffsets(timestampsToSearch.keySet());

            consumer.assign(topicPartitions);
            consumer.seekToBeginning(topicPartitions);
            final Iterator<byte[]> iter = new Iterator<byte[]>() {
                private static final int TIMEOUT_MS = 10000;
                private Iterator<ConsumerRecord<byte[], byte[]>> recordIter = null;
                private byte[] next = null;

                private byte[] fetchNext() {
                    while (true) {
                        long curTime = Time.SYSTEM.milliseconds();
                        if (curTime - prodTimeMs > TIMEOUT_MS)
                            throw new RuntimeException("Timed out after " + TIMEOUT_MS + " ms.");
                        if (recordIter == null) {
                            ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(100));
                            recordIter = records.iterator();
                        }
                        if (recordIter.hasNext())
                            return recordIter.next().value();
                        recordIter = null;
                    }
                }

                @Override
                public boolean hasNext() {
                    if (next != null)
                        return true;
                    next = fetchNext();
                    return next != null;
                }

                @Override
                public byte[] next() {
                    if (!hasNext())
                        throw new NoSuchElementException();
                    byte[] cur = next;
                    next = null;
                    return cur;
                }

                @Override
                public void remove() {
                    throw new UnsupportedOperationException();
                }
            };
            byte[] next = iter.next();
            try {
                compareArrays(message1, next);
                log.debug("Found first message...");
            } catch (RuntimeException e) {
                throw new RuntimeException("The first message in this topic was not ours. Please use a new topic when " +
                        "running this program.");
            }
            try {
                next = iter.next();
                if (testConfig.expectRecordTooLargeException) {
                    throw new RuntimeException("Expected to get a RecordTooLargeException when reading a record " +
                            "bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG);
                }
                try {
                    compareArrays(message2, next);
                } catch (RuntimeException e) {
                    System.out.println("The second message in this topic was not ours. Please use a new " +
                        "topic when running this program.");
                    Exit.exit(1);
                }
            } catch (RecordTooLargeException e) {
                log.debug("Got RecordTooLargeException", e);
                if (!testConfig.expectRecordTooLargeException)
                    throw new RuntimeException("Got an unexpected RecordTooLargeException when reading a record " +
                        "bigger than " + ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG);
            }
            log.debug("Closing consumer.");
        }
        log.info("Closed consumer.");
    }
ClientCompatibilityTest is invoked by the ducktape test client_compatibility_features_test.py to validate
client behavior when various broker versions are in use.  It runs various client operations and tests whether they
are supported or not.
tools/src/main/java/org/apache/kafka/tools/ClientCompatibilityTest.java