content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.connect.integration;
  
  import org.apache.kafka.clients.admin.Admin;
  import org.apache.kafka.clients.admin.NewTopic;
  import org.apache.kafka.clients.consumer.ConsumerConfig;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.producer.KafkaProducer;
  import org.apache.kafka.clients.producer.Producer;
  import org.apache.kafka.clients.producer.ProducerRecord;
  import org.apache.kafka.common.acl.AccessControlEntry;
  import org.apache.kafka.common.acl.AclBinding;
  import org.apache.kafka.common.acl.AclOperation;
  import org.apache.kafka.common.acl.AclPermissionType;
  import org.apache.kafka.common.config.ConfigDef;
  import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
  import org.apache.kafka.common.errors.InvalidProducerEpochException;
  import org.apache.kafka.common.resource.PatternType;
  import org.apache.kafka.common.resource.ResourcePattern;
  import org.apache.kafka.common.resource.ResourceType;
  import org.apache.kafka.connect.connector.Task;
  import org.apache.kafka.connect.errors.ConnectException;
  import org.apache.kafka.connect.json.JsonConverter;
  import org.apache.kafka.connect.json.JsonConverterConfig;
  import org.apache.kafka.connect.runtime.Worker;
  import org.apache.kafka.connect.runtime.WorkerConfig;
  import org.apache.kafka.connect.runtime.distributed.DistributedConfig;
  import org.apache.kafka.connect.runtime.rest.entities.ConfigInfo;
  import org.apache.kafka.connect.runtime.rest.entities.ConfigInfos;
  import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo;
  import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException;
  import org.apache.kafka.connect.source.SourceConnector;
  import org.apache.kafka.connect.source.SourceRecord;
  import org.apache.kafka.connect.source.SourceTask;
  import org.apache.kafka.connect.storage.StringConverter;
  import org.apache.kafka.connect.util.clusters.ConnectAssertions;
  import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster;
  import org.apache.kafka.connect.util.clusters.EmbeddedKafkaCluster;
  import org.apache.kafka.network.SocketServerConfigs;
  import org.apache.kafka.server.config.KRaftConfigs;
  import org.apache.kafka.server.config.ServerConfigs;
  import org.apache.kafka.test.NoRetryException;
  
  import org.junit.jupiter.api.AfterEach;
  import org.junit.jupiter.api.BeforeEach;
  import org.junit.jupiter.api.Tag;
  import org.junit.jupiter.api.Test;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Closeable;
  import java.time.Duration;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Properties;
  import java.util.Set;
  import java.util.concurrent.TimeUnit;
  import java.util.stream.Collectors;
  import java.util.stream.IntStream;
  import java.util.stream.LongStream;
  
  import static org.apache.kafka.clients.producer.ProducerConfig.BOOTSTRAP_SERVERS_CONFIG;
  import static org.apache.kafka.clients.producer.ProducerConfig.CLIENT_ID_CONFIG;
  import static org.apache.kafka.clients.producer.ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG;
  import static org.apache.kafka.clients.producer.ProducerConfig.TRANSACTIONAL_ID_CONFIG;
  import static org.apache.kafka.connect.integration.TestableSourceConnector.CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG;
  import static org.apache.kafka.connect.integration.TestableSourceConnector.CUSTOM_TRANSACTION_BOUNDARIES_CONFIG;
  import static org.apache.kafka.connect.integration.TestableSourceConnector.MAX_MESSAGES_PER_SECOND_CONFIG;
  import static org.apache.kafka.connect.integration.TestableSourceConnector.MESSAGES_PER_POLL_CONFIG;
  import static org.apache.kafka.connect.integration.TestableSourceConnector.TOPIC_CONFIG;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLASS_CONFIG;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.KEY_CONVERTER_CLASS_CONFIG;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.NAME_CONFIG;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.TASKS_MAX_CONFIG;
  import static org.apache.kafka.connect.runtime.ConnectorConfig.VALUE_CONVERTER_CLASS_CONFIG;
  import static org.apache.kafka.connect.runtime.SourceConnectorConfig.EXACTLY_ONCE_SUPPORT_CONFIG;
  import static org.apache.kafka.connect.runtime.SourceConnectorConfig.OFFSETS_TOPIC_CONFIG;
  import static org.apache.kafka.connect.runtime.SourceConnectorConfig.TRANSACTION_BOUNDARY_CONFIG;
  import static org.apache.kafka.connect.runtime.SourceConnectorConfig.TRANSACTION_BOUNDARY_INTERVAL_CONFIG;
  import static org.apache.kafka.connect.runtime.distributed.DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG;
  import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.CONNECTOR;
  import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.INTERVAL;
  import static org.apache.kafka.connect.source.SourceTask.TransactionBoundary.POLL;
  import static org.apache.kafka.test.TestUtils.waitForCondition;
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertFalse;
  import static org.junit.jupiter.api.Assertions.assertInstanceOf;
  import static org.junit.jupiter.api.Assertions.assertNotNull;
  import static org.junit.jupiter.api.Assertions.assertThrows;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  @Tag("integration")
  public class ExactlyOnceSourceIntegrationTest {
  
      private static final Logger log = LoggerFactory.getLogger(ExactlyOnceSourceIntegrationTest.class);
      private static final String CLUSTER_GROUP_ID = "exactly-once-source-integration-test";
      private static final String CONNECTOR_NAME = "exactlyOnceQuestionMark";
  
      private static final int CONSUME_RECORDS_TIMEOUT_MS = 60_000;
      private static final int SOURCE_TASK_PRODUCE_TIMEOUT_MS = 30_000;
      private static final int ACL_PROPAGATION_TIMEOUT_MS = 30_000;
      private static final int DEFAULT_NUM_WORKERS = 3;
  
      // Tests require that a minimum but not unreasonably large number of records are sourced.
      // Throttle the poll such that a reasonable amount of records are produced while the test runs.
      private static final int MINIMUM_MESSAGES = 100;
      private static final String MESSAGES_PER_POLL = Integer.toString(MINIMUM_MESSAGES);
      private static final String MESSAGES_PER_SECOND = Long.toString(MINIMUM_MESSAGES / 2);
  
      private Properties brokerProps;
      private Map<String, String> workerProps;
      private EmbeddedConnectCluster.Builder connectBuilder;
      private EmbeddedConnectCluster connect;
      private ConnectorHandle connectorHandle;
  
      @BeforeEach
      public void setup() {
          workerProps = new HashMap<>();
          workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "enabled");
          workerProps.put(DistributedConfig.GROUP_ID_CONFIG, CLUSTER_GROUP_ID);
  
          brokerProps = new Properties();
          brokerProps.put("transaction.state.log.replication.factor", "1");
          brokerProps.put("transaction.state.log.min.isr", "1");
  
          // build a Connect cluster backed by Kafka
          connectBuilder = new EmbeddedConnectCluster.Builder()
                  .numWorkers(DEFAULT_NUM_WORKERS)
                  .numBrokers(1)
                  .workerProps(workerProps)
                  .brokerProps(brokerProps);
  
          // get a handle to the connector
          connectorHandle = RuntimeHandles.get().connectorHandle(CONNECTOR_NAME);
      }
  
      private void startConnect() {
          connect = connectBuilder.build();
          connect.start();
      }
  
      @AfterEach
      public void close() {
          try {
              // stop the Connect cluster and its backing Kafka cluster.
              connect.stop();
          } finally {
              // Clear the handle for the connector. Fun fact: if you don't do this, your tests become quite flaky.
              RuntimeHandles.get().deleteConnector(CONNECTOR_NAME);
          }
      }
  
      /**
       * A simple test for the pre-flight validation API for connectors to provide their own guarantees for exactly-once semantics.
       */
      @Test
      public void testPreflightValidation() {
          connectBuilder.numWorkers(1);
          startConnect();
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, "1");
          props.put(TOPIC_CONFIG, "topic");
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
  
          // Test out the "exactly.once.support" property
          props.put(EXACTLY_ONCE_SUPPORT_CONFIG, "required");
  
          // Connector will return null from SourceConnector::exactlyOnceSupport
          props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, TestableSourceConnector.EXACTLY_ONCE_NULL);
          ConfigInfos validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(),
                  "Preflight validation should have exactly one error");
          ConfigInfo propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(),
                  "Preflight validation for exactly-once support property should have at least one error message");
  
          // Connector will return UNSUPPORTED from SourceConnector::exactlyOnceSupport
          props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, TestableSourceConnector.EXACTLY_ONCE_UNSUPPORTED);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
          propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(), 
                  "Preflight validation for exactly-once support property should have at least one error message");
  
          // Connector will throw an exception from SourceConnector::exactlyOnceSupport
          props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, TestableSourceConnector.EXACTLY_ONCE_FAIL);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
          propertyValidation = findConfigInfo(EXACTLY_ONCE_SUPPORT_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(),
                  "Preflight validation for exactly-once support property should have at least one error message");
  
          // Connector will return SUPPORTED from SourceConnector::exactlyOnceSupport
          props.put(CUSTOM_EXACTLY_ONCE_SUPPORT_CONFIG, TestableSourceConnector.EXACTLY_ONCE_SUPPORTED);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(0, validation.errorCount(), "Preflight validation should have zero errors");
  
          // Test out the transaction boundary definition property
          props.put(TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());
  
          // Connector will return null from SourceConnector::canDefineTransactionBoundaries
          props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, TestableSourceConnector.TRANSACTION_BOUNDARIES_NULL);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
          propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(),
                  "Preflight validation for transaction boundary property should have at least one error message");
  
          // Connector will return UNSUPPORTED from SourceConnector::canDefineTransactionBoundaries
          props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, TestableSourceConnector.TRANSACTION_BOUNDARIES_UNSUPPORTED);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
          propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(),
                  "Preflight validation for transaction boundary property should have at least one error message");
  
          // Connector will throw an exception from SourceConnector::canDefineTransactionBoundaries
          props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, TestableSourceConnector.TRANSACTION_BOUNDARIES_FAIL);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(1, validation.errorCount(), "Preflight validation should have exactly one error");
          propertyValidation = findConfigInfo(TRANSACTION_BOUNDARY_CONFIG, validation);
          assertFalse(propertyValidation.configValue().errors().isEmpty(), 
                  "Preflight validation for transaction boundary property should have at least one error message");
  
          // Connector will return SUPPORTED from SourceConnector::canDefineTransactionBoundaries
          props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, TestableSourceConnector.TRANSACTION_BOUNDARIES_SUPPORTED);
          validation = connect.validateConnectorConfig(TestableSourceConnector.class.getSimpleName(), props);
          assertEquals(0, validation.errorCount(), "Preflight validation should have zero errors");
      }
  
      /**
       * A simple green-path test that ensures the worker can start up a source task with exactly-once support enabled
       * and write some records to Kafka that will be visible to a downstream consumer using the "READ_COMMITTED"
       * isolation level. The "poll" transaction boundary is used.
       */
      @Test
      public void testPollBoundary() throws Exception {
          // Much slower offset commit interval; should never be triggered during this test
          workerProps.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "600000");
          connectBuilder.numWorkers(1);
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          int numTasks = 1;
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TRANSACTION_BOUNDARY_CONFIG, POLL.toString());
          props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
          props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
  
          // expect all records to be consumed and committed by the connector
          connectorHandle.expectedRecords(MINIMUM_MESSAGES);
          connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
          // start a source connector
          connect.configureConnector(CONNECTOR_NAME, props);
  
          log.info("Waiting for records to be provided to worker by task");
          // wait for the connector tasks to produce enough records
          connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
          log.info("Waiting for records to be committed to Kafka by worker");
          // wait for the connector tasks to commit enough records
          connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
          StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
          connect.deleteConnector(CONNECTOR_NAME);
          assertConnectorStopped(connectorStop);
  
          // consume all records from the source topic or fail, to ensure that they were correctly produced
          ConsumerRecords<byte[], byte[]> records = connect.kafka().consumeAll(
                  CONSUME_RECORDS_TIMEOUT_MS,
                  Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                  null,
                  topic
          );
          assertTrue(records.count() >= MINIMUM_MESSAGES,
                  "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + records.count());
          assertExactlyOnceSeqnos(records, numTasks);
      }
  
      /**
       * A simple green-path test that ensures the worker can start up a source task with exactly-once support enabled
       * and write some records to Kafka that will be visible to a downstream consumer using the "READ_COMMITTED"
       * isolation level. The "interval" transaction boundary is used with a connector-specific override.
       */
      @Test
      public void testIntervalBoundary() throws Exception {
          // Much slower offset commit interval; should never be triggered during this test
          workerProps.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "600000");
          connectBuilder.numWorkers(1);
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          int numTasks = 1;
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TRANSACTION_BOUNDARY_CONFIG, INTERVAL.toString());
          props.put(TRANSACTION_BOUNDARY_INTERVAL_CONFIG, "10000");
          props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
          props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
  
          // expect all records to be consumed and committed by the connector
          connectorHandle.expectedRecords(MINIMUM_MESSAGES);
          connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
          // start a source connector
          connect.configureConnector(CONNECTOR_NAME, props);
  
          log.info("Waiting for records to be provided to worker by task");
          // wait for the connector tasks to produce enough records
          connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
          log.info("Waiting for records to be committed to Kafka by worker");
          // wait for the connector tasks to commit enough records
          connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
          StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
          connect.deleteConnector(CONNECTOR_NAME);
          assertConnectorStopped(connectorStop);
  
          // consume all records from the source topic or fail, to ensure that they were correctly produced
          ConsumerRecords<byte[], byte[]> records = connect.kafka().consumeAll(
                  CONSUME_RECORDS_TIMEOUT_MS,
                  Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                  null,
                  topic
          );
          assertTrue(records.count() >= MINIMUM_MESSAGES,
                  "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + records.count());
          assertExactlyOnceSeqnos(records, numTasks);
      }
  
      /**
       * A simple green-path test that ensures the worker can start up a source task with exactly-once support enabled
       * and write some records to Kafka that will be visible to a downstream consumer using the "READ_COMMITTED"
       * isolation level. The "connector" transaction boundary is used with a connector that defines transactions whose
       * size correspond to successive elements of the Fibonacci sequence, where transactions with an even number of
       * records are aborted, and those with an odd number of records are committed.
       */
      @Test
      public void testConnectorBoundary() throws Exception {
          String offsetsTopic = "exactly-once-source-cluster-offsets";
          workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, offsetsTopic);
          connectBuilder.numWorkers(1);
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, "1");
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TRANSACTION_BOUNDARY_CONFIG, CONNECTOR.toString());
          props.put(CUSTOM_TRANSACTION_BOUNDARIES_CONFIG, TestableSourceConnector.TRANSACTION_BOUNDARIES_SUPPORTED);
          props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
          props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
  
          // the connector aborts some transactions, which causes records that it has emitted (and for which
          // SourceTask::commitRecord has been invoked) to be invisible to consumers; we expect the task to
          // emit at most 233 records in total before 100 records have been emitted as part of one or more
          // committed transactions
          connectorHandle.expectedRecords(233);
          connectorHandle.expectedCommits(233);
  
          // start a source connector
          connect.configureConnector(CONNECTOR_NAME, props);
  
          log.info("Waiting for records to be provided to worker by task");
          // wait for the connector tasks to produce enough records
          connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
          log.info("Waiting for records to be committed to Kafka by worker");
          // wait for the connector tasks to commit enough records
          connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
          Map<String, Object> consumerProps = new HashMap<>();
          consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
          // consume all records from the source topic or fail, to ensure that they were correctly produced
          ConsumerRecords<byte[], byte[]> sourceRecords = connect.kafka().consumeAll(
              CONSUME_RECORDS_TIMEOUT_MS,
              Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
              null,
              topic
          );
          assertTrue(sourceRecords.count() >= MINIMUM_MESSAGES,
                  "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + sourceRecords.count());
  
          // also consume from the cluster's offsets topic to verify that the expected offsets (which should correspond to the connector's
          // custom transaction boundaries) were committed
          List<Long> expectedOffsetSeqnos = new ArrayList<>();
          long lastExpectedOffsetSeqno = 1;
          long nextExpectedOffsetSeqno = 1;
          while (nextExpectedOffsetSeqno <= MINIMUM_MESSAGES) {
              expectedOffsetSeqnos.add(nextExpectedOffsetSeqno);
              nextExpectedOffsetSeqno += lastExpectedOffsetSeqno;
              lastExpectedOffsetSeqno = nextExpectedOffsetSeqno - lastExpectedOffsetSeqno;
          }
          ConsumerRecords<byte[], byte[]> offsetRecords = connect.kafka()
                  .consume(
                          expectedOffsetSeqnos.size(),
                          TimeUnit.MINUTES.toMillis(1),
                          consumerProps,
                          offsetsTopic
                  );
  
          List<Long> actualOffsetSeqnos = parseAndAssertOffsetsForSingleTask(offsetRecords);
  
          assertEquals(expectedOffsetSeqnos, actualOffsetSeqnos.subList(0, expectedOffsetSeqnos.size()),
                  "Committed offsets should match connector-defined transaction boundaries");
  
          List<Long> expectedRecordSeqnos = LongStream.range(1, MINIMUM_MESSAGES + 1).boxed().collect(Collectors.toList());
          long priorBoundary = 1;
          long nextBoundary = 2;
          while (priorBoundary < expectedRecordSeqnos.get(expectedRecordSeqnos.size() - 1)) {
              if (nextBoundary % 2 == 0) {
                  for (long i = priorBoundary + 1; i < nextBoundary + 1; i++) {
                      expectedRecordSeqnos.remove(i);
                  }
              }
              nextBoundary += priorBoundary;
              priorBoundary = nextBoundary - priorBoundary;
          }
          List<Long> actualRecordSeqnos = parseAndAssertValuesForSingleTask(sourceRecords);
          // Have to sort the records by seqno since we produce to multiple partitions and in-order consumption isn't guaranteed
          Collections.sort(actualRecordSeqnos);
          assertEquals(expectedRecordSeqnos, actualRecordSeqnos.subList(0, expectedRecordSeqnos.size()),
                  "Committed records should exclude connector-aborted transactions");
      }
  
      /**
       * Brings up a one-node cluster, then intentionally fences out the transactional producer used by the leader
       * for writes to the config topic to simulate a zombie leader being active in the cluster. The leader should
       * automatically recover, verify that it is still the leader, and then succeed to create a connector when the
       * user resends the request.
       */
      @Test
      public void testFencedLeaderRecovery() throws Exception {
          connectBuilder.numWorkers(1);
          // Much slower offset commit interval; should never be triggered during this test
          workerProps.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "600000");
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          int numTasks = 1;
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TRANSACTION_BOUNDARY_CONFIG, POLL.toString());
          props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
          props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
  
          // expect all records to be consumed and committed by the connector
          connectorHandle.expectedRecords(MINIMUM_MESSAGES);
          connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
          // fence out the leader of the cluster
          Producer<?, ?> zombieLeader = transactionalProducer(
                  "simulated-zombie-leader",
                  DistributedConfig.transactionalProducerId(CLUSTER_GROUP_ID)
          );
          zombieLeader.initTransactions();
          zombieLeader.close();
  
          // start a source connector--should fail the first time
          assertThrows(ConnectRestException.class, () -> connect.configureConnector(CONNECTOR_NAME, props));
  
          // the second request should succeed because the leader has reclaimed write privileges for the config topic
          connect.configureConnector(CONNECTOR_NAME, props);
  
          log.info("Waiting for records to be provided to worker by task");
          // wait for the connector tasks to produce enough records
          connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
          log.info("Waiting for records to be committed to Kafka by worker");
          // wait for the connector tasks to commit enough records
          connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
          StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
          connect.deleteConnector(CONNECTOR_NAME);
          assertConnectorStopped(connectorStop);
  
          // consume all records from the source topic or fail, to ensure that they were correctly produced
          ConsumerRecords<byte[], byte[]> records = connect.kafka().consumeAll(
                  CONSUME_RECORDS_TIMEOUT_MS,
                  Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                  null,
                  topic
          );
          assertTrue(records.count() >= MINIMUM_MESSAGES,
                  "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + records.count());
          assertExactlyOnceSeqnos(records, numTasks);
      }
  
      /**
       * A moderately-complex green-path test that ensures the worker can start up and run tasks for a source
       * connector that gets reconfigured, and will fence out potential zombie tasks for older generations before
       * bringing up new task instances.
       */
      @Test
      public void testConnectorReconfiguration() throws Exception {
          // Much slower offset commit interval; should never be triggered during this test
          workerProps.put(WorkerConfig.OFFSET_COMMIT_INTERVAL_MS_CONFIG, "600000");
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          Map<String, String> props = new HashMap<>();
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
          props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
  
          // expect all records to be consumed and committed by the connector
          connectorHandle.expectedRecords(MINIMUM_MESSAGES);
          connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
          StartAndStopLatch connectorStart = connectorAndTaskStart(3);
          props.put(TASKS_MAX_CONFIG, "3");
          // start a source connector
          connect.configureConnector(CONNECTOR_NAME, props);
          assertConnectorStarted(connectorStart);
  
          assertProducersAreFencedOnReconfiguration(3, 5, topic, props);
          assertProducersAreFencedOnReconfiguration(5, 1, topic, props);
          assertProducersAreFencedOnReconfiguration(1, 5, topic, props);
          assertProducersAreFencedOnReconfiguration(5, 3, topic, props);
  
          // Do a final sanity check to make sure that the last generation of tasks is able to run
          log.info("Waiting for records to be provided to worker by task");
          // wait for the connector tasks to produce enough records
          connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
          log.info("Waiting for records to be committed to Kafka by worker");
          // wait for the connector tasks to commit enough records
          connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
          StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
          connect.deleteConnector(CONNECTOR_NAME);
          assertConnectorStopped(connectorStop);
  
          // consume all records from the source topic or fail, to ensure that they were correctly produced
          ConsumerRecords<byte[], byte[]> records = connect.kafka().consumeAll(
                  CONSUME_RECORDS_TIMEOUT_MS,
                  Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                  null,
                  topic
          );
          assertTrue(records.count() >= MINIMUM_MESSAGES,
                  "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + records.count());
          // We used at most five tasks during the tests; each of them should have been able to produce records
          assertExactlyOnceSeqnos(records, 5);
      }
  
      /**
       * This test ensures that tasks are marked failed in the status API when the round of
       * zombie fencing that takes place before they are brought up fails. In addition, once
       * the issue with the connector config that made fencing impossible is rectified, tasks
       * can be successfully restarted.
       * <p>
       * Fencing failures are induced by bringing up an ACL-secured Kafka cluster and creating
       * a connector whose principal is not authorized to access the transactional IDs that Connect
       * uses for its tasks.
       * <p>
       * When the connector is initially brought up, no fencing is necessary. However, once it is
       * reconfigured and generates new task configs, a round of zombie fencing is triggered,
       * and all of its tasks fail when that round of zombie fencing fails.
       * <p>
       * After, the connector's principal is granted access to the necessary transactional IDs,
       * all of its tasks are restarted, and we verify that they are able to come up successfully
       * this time.
       */
      @Test
      public void testTasksFailOnInabilityToFence() throws Exception {
          brokerProps.put(SocketServerConfigs.LISTENER_SECURITY_PROTOCOL_MAP_CONFIG, "CONTROLLER:SASL_PLAINTEXT,EXTERNAL:SASL_PLAINTEXT");
          brokerProps.put(ServerConfigs.AUTHORIZER_CLASS_NAME_CONFIG, "org.apache.kafka.metadata.authorizer.StandardAuthorizer");
          brokerProps.put(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG, "PLAIN");
          brokerProps.put(BrokerSecurityConfigs.SASL_MECHANISM_INTER_BROKER_PROTOCOL_CONFIG, "PLAIN");
          brokerProps.put(KRaftConfigs.SASL_MECHANISM_CONTROLLER_PROTOCOL_CONFIG, "PLAIN");
          String listenerSaslJaasConfig = "org.apache.kafka.common.security.plain.PlainLoginModule required "
                  + "username=\"super\" "
                  + "password=\"super_pwd\" "
                  + "user_connector=\"connector_pwd\" "
                  + "user_super=\"super_pwd\";";
          brokerProps.put("listener.name.external.plain.sasl.jaas.config", listenerSaslJaasConfig);
          brokerProps.put("listener.name.controller.plain.sasl.jaas.config", listenerSaslJaasConfig);
          brokerProps.put("super.users", "User:super");
  
          Map<String, String> superUserClientConfig = new HashMap<>();
          superUserClientConfig.put("sasl.mechanism", "PLAIN");
          superUserClientConfig.put("security.protocol", "SASL_PLAINTEXT");
          superUserClientConfig.put("sasl.jaas.config",
                  "org.apache.kafka.common.security.plain.PlainLoginModule required "
                          + "username=\"super\" "
                          + "password=\"super_pwd\";");
          // Give the worker super-user privileges
          workerProps.putAll(superUserClientConfig);
  
          final String globalOffsetsTopic = "connect-worker-offsets-topic";
          workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, globalOffsetsTopic);
  
          connectBuilder.clientProps(superUserClientConfig);
  
          startConnect();
  
          String topic = "test-topic";
          try (Admin admin = connect.kafka().createAdminClient()) {
              admin.createTopics(Collections.singleton(new NewTopic(topic, 3, (short) 1))).all().get();
          }
  
          Map<String, String> props = new HashMap<>();
          int tasksMax = 2; // Use two tasks since single-task connectors don't require zombie fencing
          props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
          props.put(TOPIC_CONFIG, topic);
          props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TASKS_MAX_CONFIG, Integer.toString(tasksMax));
          // Give the connectors' consumer and producer super-user privileges
          superUserClientConfig.forEach((property, value) -> {
              props.put(CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + property, value);
              props.put(CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + property, value);
          });
          // But limit its admin client's privileges
          props.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + "sasl.mechanism", "PLAIN");
          props.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + "security.protocol", "SASL_PLAINTEXT");
          props.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + "sasl.jaas.config",
                  "org.apache.kafka.common.security.plain.PlainLoginModule required "
                          + "username=\"connector\" "
                          + "password=\"connector_pwd\";");
          // Grant the connector's admin permissions to access the topics for its records and offsets
          // Intentionally leave out permissions required for fencing
          try (Admin admin = connect.kafka().createAdminClient()) {
              admin.createAcls(Arrays.asList(
                      new AclBinding(
                              new ResourcePattern(ResourceType.TOPIC, topic, PatternType.LITERAL),
                              new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                      ),
                      new AclBinding(
                              new ResourcePattern(ResourceType.TOPIC, globalOffsetsTopic, PatternType.LITERAL),
                              new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                      )
              )).all().get();
          }
  
          log.info("Bringing up connector with fresh slate; fencing should not be necessary");
          connect.configureConnector(CONNECTOR_NAME, props);
  
          // Hack: There is a small chance that our recent ACL updates for the connector have
          // not yet been propagated across the entire Kafka cluster, and that our connector
          // will fail on startup when it tries to list the end offsets of the worker's offsets topic
          // So, we implement some retry logic here to add a layer of resiliency in that case
          waitForCondition(
                  () -> {
                      ConnectorStateInfo status = connect.connectorStatus(CONNECTOR_NAME);
                      if ("RUNNING".equals(status.connector().state())) {
                          return true;
                      } else if ("FAILED".equals(status.connector().state())) {
                          log.debug("Restarting failed connector {}", CONNECTOR_NAME);
                          connect.restartConnector(CONNECTOR_NAME);
                      }
                      return false;
                  },
                  ACL_PROPAGATION_TIMEOUT_MS,
                  "Connector was not able to start in time, "
                          + "or ACL updates were not propagated across the Kafka cluster soon enough"
          );
  
          // Also verify that the connector's tasks have been able to start successfully
          connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(CONNECTOR_NAME, tasksMax, "Connector and task should have started successfully");
  
          log.info("Reconfiguring connector; fencing should be necessary, and tasks should fail to start");
          props.put("message.in.a.bottle", "19e184427ac45bd34c8588a4e771aa1a");
          connect.configureConnector(CONNECTOR_NAME, props);
  
          // Verify that the task has failed, and that the failure is visible to users via the REST API
          connect.assertions().assertConnectorIsRunningAndTasksHaveFailed(CONNECTOR_NAME, tasksMax, "Task should have failed on startup");
  
          // Now grant the necessary permissions for fencing to the connector's admin
          try (Admin admin = connect.kafka().createAdminClient()) {
              admin.createAcls(Arrays.asList(
                      new AclBinding(
                              new ResourcePattern(ResourceType.TRANSACTIONAL_ID, Worker.taskTransactionalId(CLUSTER_GROUP_ID, CONNECTOR_NAME, 0), PatternType.LITERAL),
                              new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                      ),
                      new AclBinding(
                              new ResourcePattern(ResourceType.TRANSACTIONAL_ID, Worker.taskTransactionalId(CLUSTER_GROUP_ID, CONNECTOR_NAME, 1), PatternType.LITERAL),
                              new AccessControlEntry("User:connector", "*", AclOperation.ALL, AclPermissionType.ALLOW)
                      )
              ));
          }
  
          log.info("Restarting connector after tweaking its ACLs; fencing should succeed this time");
          connect.restartConnectorAndTasks(CONNECTOR_NAME, false, true, false);
  
          // Verify that the connector and its tasks have been able to restart successfully
          // Use the same retry logic as above, in case there is a delay in the propagation of our ACL updates
          waitForCondition(
                  () -> {
                      ConnectorStateInfo status = connect.connectorStatus(CONNECTOR_NAME);
                      boolean connectorRunning = "RUNNING".equals(status.connector().state());
                      boolean allTasksRunning = status.tasks().stream()
                              .allMatch(t -> "RUNNING".equals(t.state()));
                      boolean expectedNumTasks = status.tasks().size() == tasksMax;
                      if (connectorRunning && allTasksRunning && expectedNumTasks) {
                          return true;
                      } else {
                          if (!connectorRunning) {
                              if ("FAILED".equals(status.connector().state())) {
                                  // Only task failures are expected ;if the connector has failed, something
                                  // else is wrong and we should fail the test immediately
                                  throw new NoRetryException(
                                          new AssertionError("Connector " + CONNECTOR_NAME + " has failed unexpectedly")
                                  );
                              }
                          }
                          // Restart all failed tasks
                          status.tasks().stream()
                                  .filter(t -> "FAILED".equals(t.state()))
                                  .map(ConnectorStateInfo.TaskState::id)
                                  .forEach(t -> connect.restartTask(CONNECTOR_NAME, t));
                          return false;
                      }
                  },
                  ConnectAssertions.CONNECTOR_SETUP_DURATION_MS,
                  "Connector and task should have restarted successfully"
          );
      }
  
      /**
       * This test focuses extensively on the per-connector offsets feature.
       * <p>
       * First, a connector is brought up whose producer is configured to write to a different Kafka cluster
       * than the one the Connect cluster users for its internal topics, then the contents of the connector's
       * dedicated offsets topic and the worker's internal offsets topic are inspected to ensure that offsets
       * have been backed up from the dedicated topic to the global topic.
       * <p>
       * Then, a "soft downgrade" is simulated: the Connect cluster is shut down and reconfigured to disable
       * exactly-once support. The cluster is brought up again, the connector is allowed to produce some data,
       * the connector is shut down, and this time, the records the connector has produced are inspected for
       * accuracy. Because of the downgrade, exactly-once semantics are lost, but we check to make sure that
       * the task has maintained exactly-once semantics <i>up to the last-committed record</i>.
       */
      @Test
      public void testSeparateOffsetsTopic() throws Exception {
          final String globalOffsetsTopic = "connect-worker-offsets-topic";
          workerProps.put(DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG, globalOffsetsTopic);
  
          startConnect();
  
          int numConnectorTargetedBrokers = 1;
          EmbeddedKafkaCluster connectorTargetedCluster = new EmbeddedKafkaCluster(numConnectorTargetedBrokers, brokerProps);
          try (Closeable clusterShutdown = connectorTargetedCluster::stop) {
              connectorTargetedCluster.start();
              // Wait for the connector-targeted Kafka cluster to get on its feet
              waitForCondition(
                      () -> connectorTargetedCluster.runningBrokers().size() == numConnectorTargetedBrokers,
                      ConnectAssertions.WORKER_SETUP_DURATION_MS,
                      "Separate Kafka cluster did not start in time"
              );
  
              String topic = "test-topic";
              connectorTargetedCluster.createTopic(topic, 3);
  
              int numTasks = 1;
  
              Map<String, String> props = new HashMap<>();
              props.put(CONNECTOR_CLASS_CONFIG, TestableSourceConnector.class.getName());
              props.put(TASKS_MAX_CONFIG, Integer.toString(numTasks));
              props.put(TOPIC_CONFIG, topic);
              props.put(KEY_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
              props.put(VALUE_CONVERTER_CLASS_CONFIG, StringConverter.class.getName());
              props.put(NAME_CONFIG, CONNECTOR_NAME);
              props.put(TRANSACTION_BOUNDARY_CONFIG, POLL.toString());
              props.put(MESSAGES_PER_POLL_CONFIG, MESSAGES_PER_POLL);
              props.put(MAX_MESSAGES_PER_SECOND_CONFIG, MESSAGES_PER_SECOND);
              props.put(CONNECTOR_CLIENT_PRODUCER_OVERRIDES_PREFIX + BOOTSTRAP_SERVERS_CONFIG, connectorTargetedCluster.bootstrapServers());
              props.put(CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + BOOTSTRAP_SERVERS_CONFIG, connectorTargetedCluster.bootstrapServers());
              props.put(CONNECTOR_CLIENT_ADMIN_OVERRIDES_PREFIX + BOOTSTRAP_SERVERS_CONFIG, connectorTargetedCluster.bootstrapServers());
              String offsetsTopic = CONNECTOR_NAME + "-offsets";
              props.put(OFFSETS_TOPIC_CONFIG, offsetsTopic);
  
              // expect all records to be consumed and committed by the connector
              connectorHandle.expectedRecords(MINIMUM_MESSAGES);
              connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
              // start a source connector
              connect.configureConnector(CONNECTOR_NAME, props);
              connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                      CONNECTOR_NAME,
                      numTasks,
                      "connector and tasks did not start in time"
              );
  
              log.info("Waiting for records to be provided to worker by task");
              // wait for the connector tasks to produce enough records
              connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
              log.info("Waiting for records to be committed to Kafka by worker");
              // wait for the connector tasks to commit enough records
              connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
              // consume at least the expected number of records from the source topic or fail, to ensure that they were correctly produced
              int recordNum = connectorTargetedCluster
                      .consume(
                          MINIMUM_MESSAGES,
                              TimeUnit.MINUTES.toMillis(1),
                              Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                              "test-topic")
                      .count();
              assertTrue(recordNum >= MINIMUM_MESSAGES,
                      "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + recordNum);
  
              // also consume from the connector's dedicated offsets topic
              ConsumerRecords<byte[], byte[]> offsetRecords = connectorTargetedCluster
                      .consumeAll(
                              TimeUnit.MINUTES.toMillis(1),
                              Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                              null,
                              offsetsTopic
                      );
              List<Long> seqnos = parseAndAssertOffsetsForSingleTask(offsetRecords);
              seqnos.forEach(seqno ->
                  assertEquals(0, seqno % MINIMUM_MESSAGES,
                          "Offset commits should occur on connector-defined poll boundaries, which happen every " + MINIMUM_MESSAGES + " records")
              );
  
              // also consume from the cluster's global offsets topic
              offsetRecords = connect.kafka().consumeAll(TimeUnit.MINUTES.toMillis(1), globalOffsetsTopic);
              seqnos = parseAndAssertOffsetsForSingleTask(offsetRecords);
              seqnos.forEach(seqno ->
                  assertEquals(0, seqno % MINIMUM_MESSAGES,
                          "Offset commits should occur on connector-defined poll boundaries, which happen every " + MINIMUM_MESSAGES + " records")
              );
  
              // Shut down the whole cluster
              connect.workers().forEach(connect::removeWorker);
              // Reconfigure the cluster with exactly-once support disabled
              workerProps.put(EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG, "disabled");
  
              // Establish new expectations for records+offsets
              connectorHandle.expectedRecords(MINIMUM_MESSAGES);
              connectorHandle.expectedCommits(MINIMUM_MESSAGES);
  
              // Restart the whole cluster
              for (int i = 0; i < DEFAULT_NUM_WORKERS; i++) {
                  connect.addWorker();
              }
  
              // And perform a basic sanity check that the cluster is able to come back up, our connector and its task are able to resume running,
              // and the task is still able to produce source records and commit offsets
              connect.assertions().assertAtLeastNumWorkersAreUp(DEFAULT_NUM_WORKERS, "cluster did not restart in time");
              connect.assertions().assertConnectorAndExactlyNumTasksAreRunning(
                      CONNECTOR_NAME,
                      1,
                      "connector and tasks did not resume running after cluster restart in time"
              );
  
              log.info("Waiting for records to be provided to worker by task");
              // wait for the connector tasks to produce enough records
              connectorHandle.awaitRecords(SOURCE_TASK_PRODUCE_TIMEOUT_MS);
  
              log.info("Waiting for records to be committed to Kafka by worker");
              // wait for the connector tasks to commit enough records
              connectorHandle.awaitCommits(TimeUnit.MINUTES.toMillis(1));
  
              StartAndStopLatch connectorStop = connectorHandle.expectedStops(1, true);
              connect.deleteConnector(CONNECTOR_NAME);
              assertConnectorStopped(connectorStop);
  
              // consume all records from the source topic or fail, to ensure that they were correctly produced
              ConsumerRecords<byte[], byte[]> sourceRecords = connectorTargetedCluster.consumeAll(
                      CONSUME_RECORDS_TIMEOUT_MS,
                      Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                      null,
                      topic
              );
              assertTrue(sourceRecords.count() >= MINIMUM_MESSAGES,
                      "Not enough records produced by source connector. Expected at least: " + MINIMUM_MESSAGES + " + but got " + sourceRecords.count());
              // also have to check which offsets have actually been committed, since we no longer have exactly-once semantics
              offsetRecords = connectorTargetedCluster.consumeAll(
                      CONSUME_RECORDS_TIMEOUT_MS,
                      Collections.singletonMap(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed"),
                      null,
                      offsetsTopic
              );
              assertAtLeastOnceSeqnos(sourceRecords, offsetRecords, numTasks);
          }
      }
  
      /**
       * A simple test to ensure that source tasks fail when trying to produce to their own offsets topic.
       * <p>
       * We fail the tasks in order to prevent deadlock that occurs when:
       * <ol>
       *     <li>
       *         A task provides a record whose topic is the task's offsets topic
       *     </li>
       *     <li>
       *         That record is dispatched to the task's producer in a transaction that remains open
       *         at least until the worker polls the task again
       *     </li>
       *     <li>
       *         In the subsequent call to SourceTask::poll, the task requests offsets from the worker
       *         (which requires a read to the end of the offsets topic, and will block until any open
       *         transactions on the topic are either committed or aborted)
       *     </li>
       * </ol>
       */
      @Test
      public void testPotentialDeadlockWhenProducingToOffsetsTopic() throws Exception {
          connectBuilder.numWorkers(1);
          startConnect();
  
          String topic = "test-topic";
          connect.kafka().createTopic(topic, 3);
  
          Map<String, String> props = new HashMap<>();
          // See below; this connector does nothing except request offsets from the worker in SourceTask::poll
          // and then return a single record targeted at its offsets topic
          props.put(CONNECTOR_CLASS_CONFIG, NaughtyConnector.class.getName());
          props.put(TASKS_MAX_CONFIG, "1");
          props.put(NAME_CONFIG, CONNECTOR_NAME);
          props.put(TRANSACTION_BOUNDARY_CONFIG, INTERVAL.toString());
          props.put(OFFSETS_TOPIC_CONFIG, "whoops");
  
          // start a source connector
          connect.configureConnector(CONNECTOR_NAME, props);
  
          connect.assertions().assertConnectorIsRunningAndTasksHaveFailed(
              CONNECTOR_NAME, 1, "Task should have failed after trying to produce to its own offsets topic");
      }
  
      private ConfigInfo findConfigInfo(String property, ConfigInfos validationResult) {
          return validationResult.values().stream()
                  .filter(info -> property.equals(info.configKey().name()))
                  .findAny()
                  .orElseThrow(() -> new AssertionError("Failed to find configuration validation result for property '" + property + "'"));
      }
  
      private List<Long> parseAndAssertOffsetsForSingleTask(ConsumerRecords<byte[], byte[]> offsetRecords) {
          Map<Integer, List<Long>> parsedOffsets = parseOffsetForTasks(offsetRecords);
          assertEquals(Collections.singleton(0), parsedOffsets.keySet(), "Expected records to only be produced from a single task");
          return parsedOffsets.get(0);
      }
  
      private List<Long> parseAndAssertValuesForSingleTask(ConsumerRecords<byte[], byte[]> sourceRecords) {
          Map<Integer, List<Long>> parsedValues = parseValuesForTasks(sourceRecords);
          assertEquals(Collections.singleton(0), parsedValues.keySet(), "Expected records to only be produced from a single task");
          return parsedValues.get(0);
      }
  
      private void assertExactlyOnceSeqnos(ConsumerRecords<byte[], byte[]> sourceRecords, int numTasks) {
          Map<Integer, List<Long>> parsedValues = parseValuesForTasks(sourceRecords);
          assertSeqnos(parsedValues, numTasks);
      }
  
      private void assertAtLeastOnceSeqnos(ConsumerRecords<byte[], byte[]> sourceRecords, ConsumerRecords<byte[], byte[]> offsetRecords, int numTasks) {
          Map<Integer, List<Long>> parsedValues = parseValuesForTasks(sourceRecords);
          Map<Integer, Long> lastCommittedValues = parseOffsetForTasks(offsetRecords)
                  .entrySet().stream().collect(Collectors.toMap(
                          Map.Entry::getKey,
                          e -> Collections.max(e.getValue())
                  ));
          parsedValues.replaceAll((task, values) -> {
              Long committedValue = lastCommittedValues.get(task);
              assertNotNull(committedValue, "No committed offset found for task " + task);
              return values.stream().filter(v -> v <= committedValue).collect(Collectors.toList());
          });
          assertSeqnos(parsedValues, numTasks);
      }
  
      private void assertSeqnos(Map<Integer, List<Long>> parsedValues, int numTasks) {
          Set<Integer> expectedKeys = IntStream.range(0, numTasks).boxed().collect(Collectors.toSet());
          assertEquals(expectedKeys, parsedValues.keySet(), "Expected records to be produced by each task");
  
          parsedValues.forEach((taskId, seqnos) -> {
              // We don't check for order here because the records may have been produced to multiple topic partitions,
              // which makes in-order consumption impossible
              Set<Long> expectedSeqnos = LongStream.range(1, seqnos.size() + 1).boxed().collect(Collectors.toSet());
              Set<Long> actualSeqnos = new HashSet<>(seqnos);
  
              Set<Long> missingSeqnos = new HashSet<>(expectedSeqnos);
              missingSeqnos.removeAll(actualSeqnos);
              Set<Long> extraSeqnos = new HashSet<>(actualSeqnos);
              extraSeqnos.removeAll(expectedSeqnos);
  
              // Try to provide the most friendly error message possible if this test fails
              assertTrue(
                      missingSeqnos.isEmpty() && extraSeqnos.isEmpty(),
                      "Seqnos for task " + taskId + " should start at 1 and increase strictly by 1 with each record, " +
                              "but the actual seqnos did not.\n" +
                              "Seqnos that should have been emitted but were not: " + missingSeqnos + "\n" +
                              "seqnos that should not have been emitted but were: " + extraSeqnos
              );
          });
      }
  
      private Map<Integer, List<Long>> parseValuesForTasks(ConsumerRecords<byte[], byte[]> sourceRecords) {
          Map<Integer, List<Long>> result = new HashMap<>();
          for (ConsumerRecord<byte[], byte[]> sourceRecord : sourceRecords) {
              assertNotNull(sourceRecord.key(), "Record key should not be null");
              assertNotNull(sourceRecord.value(), "Record value should not be null");
  
              String key = new String(sourceRecord.key());
              String value = new String(sourceRecord.value());
  
              String keyPrefix = "key-";
              String valuePrefix = "value-";
  
              assertTrue(key.startsWith(keyPrefix), "Key should start with \"" + keyPrefix + "\"");
              assertTrue(value.startsWith(valuePrefix), "Value should start with \"" + valuePrefix + "\"");
              assertEquals(
                      key.substring(keyPrefix.length()),
                      value.substring(valuePrefix.length()),
                      "key and value should be identical after prefix"
              );
  
              String[] split = key.substring(keyPrefix.length()).split("-");
              assertEquals(3, split.length, "Key should match pattern 'key-<connectorName>-<taskId>-<seqno>");
              assertEquals(CONNECTOR_NAME, split[0], "Key should match pattern 'key-<connectorName>-<taskId>-<seqno>");
  
              int taskId;
              try {
                  taskId = Integer.parseInt(split[1], 10);
              } catch (NumberFormatException e) {
                  throw new AssertionError("Task ID in key should be an integer, was '" + split[1] + "'", e);
              }
  
              long seqno;
              try {
                  seqno = Long.parseLong(split[2], 10);
              } catch (NumberFormatException e) {
                  throw new AssertionError("Seqno in key should be a long, was '" + split[2] + "'", e);
              }
  
              result.computeIfAbsent(taskId, t -> new ArrayList<>()).add(seqno);
          }
          return result;
      }
  
      private Map<Integer, List<Long>> parseOffsetForTasks(ConsumerRecords<byte[], byte[]> offsetRecords) {
          JsonConverter offsetsConverter = new JsonConverter();
          // The JSON converter behaves identically for keys and values. If that ever changes, we may need to update this test to use
          // separate converter instances.
          offsetsConverter.configure(Collections.singletonMap(JsonConverterConfig.SCHEMAS_ENABLE_CONFIG, "false"), false);
  
          Map<Integer, List<Long>> result = new HashMap<>();
          for (ConsumerRecord<byte[], byte[]> offsetRecord : offsetRecords) {
              Object keyObject = offsetsConverter.toConnectData("topic name is not used by converter", offsetRecord.key()).value();
              Object valueObject = offsetsConverter.toConnectData("topic name is not used by converter", offsetRecord.value()).value();
  
              assertNotNull(keyObject, "Offset key should not be null");
              assertNotNull(valueObject, "Offset value should not be null");
  
              @SuppressWarnings("unchecked")
              List<Object> key = assertAndCast(keyObject, List.class, "Key");
              assertEquals(
                      2,
                      key.size(),
                      "Offset topic key should be a list containing two elements: the name of the connector, and the connector-provided source partition"
              );
              assertEquals(CONNECTOR_NAME, key.get(0));
              @SuppressWarnings("unchecked")
              Map<String, Object> partition = assertAndCast(key.get(1), Map.class, "Key[1]");
              Object taskIdObject = partition.get("task.id");
              assertNotNull(taskIdObject, "Serialized source partition should contain 'task.id' field from TestableSourceConnector");
              String taskId = assertAndCast(taskIdObject, String.class, "task ID");
              assertTrue(taskId.startsWith(CONNECTOR_NAME + "-"), "task ID should match pattern '<connectorName>-<taskId>");
              String taskIdRemainder = taskId.substring(CONNECTOR_NAME.length() + 1);
              int taskNum;
              try {
                  taskNum = Integer.parseInt(taskIdRemainder);
              } catch (NumberFormatException e) {
                  throw new AssertionError("task ID should match pattern '<connectorName>-<taskId>', where <taskId> is an integer", e);
              }
  
              @SuppressWarnings("unchecked")
              Map<String, Object> value = assertAndCast(valueObject, Map.class, "Value");
  
              Object seqnoObject = value.get("saved");
              assertNotNull(seqnoObject, "Serialized source offset should contain 'seqno' field from TestableSourceConnector");
              long seqno = assertAndCast(seqnoObject, Long.class, "Seqno offset field");
  
              result.computeIfAbsent(taskNum, t -> new ArrayList<>()).add(seqno);
          }
          return result;
      }
  
      @SuppressWarnings("unchecked")
      private static <T> T assertAndCast(Object o, Class<T> klass, String objectDescription) {
          String className = o == null ? "null" : o.getClass().getName();
          assertInstanceOf(klass, o, objectDescription + " should be " + klass.getName() + "; was " + className + " instead");
          return (T) o;
      }
  
      /**
       * Clear all existing task handles for the connector, then preemptively create {@code numTasks} many task handles for it,
       * and return a {@link StartAndStopLatch} that can be used to {@link StartAndStopLatch#await(long, TimeUnit) await}
       * the startup of that connector and the expected number of tasks.
       * @param numTasks the number of tasks that should be started
       * @return a {@link StartAndStopLatch} that will block until the connector and the expected number of tasks have started
       */
      private StartAndStopLatch connectorAndTaskStart(int numTasks) {
          connectorHandle.clearTasks();
          IntStream.range(0, numTasks)
                  .mapToObj(i -> TestableSourceConnector.taskId(CONNECTOR_NAME, i))
                  .forEach(connectorHandle::taskHandle);
          return connectorHandle.expectedStarts(1, true);
      }
  
      private void assertConnectorStarted(StartAndStopLatch connectorStart) throws InterruptedException {
          assertTrue(
                  connectorStart.await(
                          ConnectAssertions.CONNECTOR_SETUP_DURATION_MS,
                          TimeUnit.MILLISECONDS
                  ),
                  "Connector and tasks did not finish startup in time"
          );
      }
  
      private void assertConnectorStopped(StartAndStopLatch connectorStop) throws InterruptedException {
          assertTrue(
                  connectorStop.await(
                          ConnectAssertions.CONNECTOR_SHUTDOWN_DURATION_MS,
                          TimeUnit.MILLISECONDS
                  ),
                  "Connector and tasks did not finish shutdown in time"
          );
      }
  
      private void assertProducersAreFencedOnReconfiguration(
              int currentNumTasks,
              int newNumTasks,
              String topic,
              Map<String, String> baseConnectorProps) throws InterruptedException {
  
          // create a collection of producers that simulate the producers used for the existing tasks
          List<KafkaProducer<byte[], byte[]>> producers = IntStream.range(0, currentNumTasks)
                  .mapToObj(i -> transactionalProducer(
                          "simulated-task-producer-" + CONNECTOR_NAME + "-" + i,
                          Worker.taskTransactionalId(CLUSTER_GROUP_ID, CONNECTOR_NAME, i)
                  )).toList();
  
          producers.forEach(KafkaProducer::initTransactions);
  
          // reconfigure the connector with a new number of tasks
          StartAndStopLatch connectorStart = connectorAndTaskStart(newNumTasks);
          baseConnectorProps.put(TASKS_MAX_CONFIG, Integer.toString(newNumTasks));
          log.info("Reconfiguring connector from {} tasks to {}", currentNumTasks, newNumTasks);
          connect.configureConnector(CONNECTOR_NAME, baseConnectorProps);
          assertConnectorStarted(connectorStart);
  
          // validate that the old producers were fenced out
          producers.forEach(producer -> assertTransactionalProducerIsFenced(producer, topic));
      }
  
      private KafkaProducer<byte[], byte[]> transactionalProducer(String clientId, String transactionalId) {
          Map<String, Object> transactionalProducerProps = new HashMap<>();
          transactionalProducerProps.put(CLIENT_ID_CONFIG, clientId);
          transactionalProducerProps.put(ENABLE_IDEMPOTENCE_CONFIG, true);
          transactionalProducerProps.put(TRANSACTIONAL_ID_CONFIG, transactionalId);
          return connect.kafka().createProducer(transactionalProducerProps);
      }
  
      private void assertTransactionalProducerIsFenced(KafkaProducer<byte[], byte[]> producer, String topic) {
          producer.beginTransaction();
          assertThrows(
                  InvalidProducerEpochException.class,
                  () -> {
                      producer.send(new ProducerRecord<>(topic, new byte[] {69}, new byte[] {96}));
                      producer.commitTransaction();
                  },
                  "Producer should be fenced out"
          );
          producer.close(Duration.ZERO);
      }
  
      public static class NaughtyConnector extends SourceConnector {
          private Map<String, String> props;
  
          @Override
          public void start(Map<String, String> props) {
              this.props = props;
          }
  
          @Override
          public Class<? extends Task> taskClass() {
              return NaughtyTask.class;
          }
  
          @Override
          public List<Map<String, String>> taskConfigs(int maxTasks) {
              return IntStream.range(0, maxTasks).mapToObj(i -> props).collect(Collectors.toList());
          }
  
          @Override
          public void stop() {
          }
  
          @Override
          public ConfigDef config() {
              return new ConfigDef();
          }
  
          @Override
          public String version() {
              return "none";
          }
      }
  
      public static class NaughtyTask extends SourceTask {
          private String topic;
  
          @Override
          public void start(Map<String, String> props) {
              if (!props.containsKey(OFFSETS_TOPIC_CONFIG)) {
                  throw new ConnectException("No offsets topic");
              }
              this.topic = props.get(OFFSETS_TOPIC_CONFIG);
          }
  
          @Override
          public List<SourceRecord> poll() {
              // Request a read to the end of the offsets topic
              context.offsetStorageReader().offset(Collections.singletonMap("", null));
              // Produce a record to the offsets topic
              return Collections.singletonList(new SourceRecord(null, null, topic, null, "", null, null));
          }
  
          @Override
          public void stop() {
          }
  
          @Override
          public String version() {
              return "none";
          }
      }
  }
  