private void verifyExpectedSinkConnectorOffsets(String connectorName, String expectedTopic, int expectedPartitions,
                                                    int expectedOffset, String conditionDetails) throws InterruptedException {
        AtomicReference<ConnectorOffsets> latestOffsets = new AtomicReference<>();
        waitForCondition(
                () -> {
                    ConnectorOffsets offsets;
                    try {
                        offsets = connect.connectorOffsets(connectorName);
                    } catch (Throwable t) {
                        log.warn("Failed to list offsets for sink connector {}", connectorName, t);
                        return false;
                    }

                    latestOffsets.set(offsets);

                    if (offsets.offsets().size() != expectedPartitions) {
                        return false;
                    }
                    for (ConnectorOffset offset: offsets.offsets()) {
                        assertEquals(expectedTopic, offset.partition().get(SinkUtils.KAFKA_TOPIC_KEY));
                        if ((Integer) offset.offset().get(SinkUtils.KAFKA_OFFSET_KEY) != expectedOffset) {
                            return false;
                        }
                    }
                    return true;
                },
                OFFSET_READ_TIMEOUT_MS,
                () -> {
                    String result = conditionDetails + ". ";
                    ConnectorOffsets offsets = latestOffsets.get();
                    if (offsets == null) {
                        result += "No attempt to list the offsets for the connector ever succeeded.";
                    } else {
                        result += "Expected every committed offset to be for topic " + expectedTopic
                                + " and for " + expectedPartitions + " partition(s) of that topic to have "
                                + "a committed offset of " + expectedOffset + ". ";
                        result += "The most-recently-available offsets for the connector are: " + offsets;
                    }
                    return result;
                }
        );
    }
Verify whether the actual consumer group offsets for a sink connector match the expected offsets. The verification
is done using the <strong><em>GET /connectors/{connector}/offsets</em></strong> REST API which is repeatedly queried
until the offsets match or the {@link #OFFSET_READ_TIMEOUT_MS timeout} is reached. Note that this assumes the following:
<ol>
    <li>The sink connector is consuming from a single Kafka topic</li>
    <li>The expected offset for each partition in the topic is the same</li>
</ol>

@param connectorName the name of the sink connector whose offsets are to be verified
@param expectedTopic the name of the Kafka topic that the sink connector is consuming from
@param expectedPartitions the number of partitions that exist for the Kafka topic
@param expectedOffset the expected consumer group offset for each partition
@param conditionDetails the condition that we're waiting to achieve (for example: Sink connector should process
                        10 records)
@throws InterruptedException if the thread is interrupted while waiting for the actual offsets to match the expected offsets
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java