private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
        return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
            Properties brokerProps = new Properties();
            brokerProps.put("transaction.state.log.replication.factor", "1");
            brokerProps.put("transaction.state.log.min.isr", "1");

            // Have to declare a new map since the passed-in one may be immutable
            Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
            // Enable fast offset commits by default
            workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));

            EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
                    .name("connect-cluster")
                    .numWorkers(NUM_WORKERS)
                    .brokerProps(brokerProps)
                    .workerProps(workerPropsWithDefaults)
                    .build();

            result.start();

            return result;
        });
    }
Integration tests for Kafka Connect's connector offset management REST APIs
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java