private void verifyExpectedSourceConnectorOffsets(String connectorName, int numTasks,
                                                      int expectedOffset, String conditionDetails) throws InterruptedException {
        waitForCondition(() -> {
            ConnectorOffsets offsets = connect.connectorOffsets(connectorName);
            // The MonitorableSourceConnector has a source partition per task
            if (offsets.offsets().size() != numTasks) {
                return false;
            }
            for (ConnectorOffset offset : offsets.offsets()) {
                assertTrue(((String) offset.partition().get("task.id")).startsWith(connectorName));
                if ((Integer) offset.offset().get("saved") != expectedOffset) {
                    return false;
                }
            }
            return true;
        }, OFFSET_READ_TIMEOUT_MS, conditionDetails);
    }
Verify whether the actual offsets for a source 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 that the source connector is a
{@link MonitorableSourceConnector}

@param connectorName the name of the source connector whose offsets are to be verified
@param numTasks the number of tasks for the source connector
@param expectedOffset the expected offset for each source partition
@param conditionDetails the condition that we're waiting to achieve (for example: Source 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