@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, MonitorableSourceConnector.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 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.
connect/runtime/src/test/java/org/apache/kafka/connect/integration/ExactlyOnceSourceIntegrationTest.java