Map<String, Config> describeTopicConfigs(Set<String> topics)
            throws InterruptedException, ExecutionException {
        Set<ConfigResource> resources = topics.stream()
            .map(x -> new ConfigResource(ConfigResource.Type.TOPIC, x))
            .collect(Collectors.toSet());
        return adminCall(
                () -> sourceAdminClient.describeConfigs(resources).all().get().entrySet().stream()
                        .collect(Collectors.toMap(x -> x.getKey().name(), Entry::getValue)),
                () -> String.format("describe configs for topics %s on %s cluster", topics, config.sourceClusterAlias())
        );
    }
@Test
    public void testMissingDescribeConfigsAcl() throws Exception {
        Admin sourceAdmin = mock(Admin.class);
        MirrorSourceConnector connector = new MirrorSourceConnector(
                sourceAdmin,
                mock(Admin.class),
                new MirrorSourceConfig(makeProps())
        );

        ExecutionException describeConfigsFailure = new ExecutionException(
                "Failed to describe topic configs",
                new TopicAuthorizationException("Topic authorization failed")
        );
        @SuppressWarnings("unchecked")
        KafkaFuture<Map<ConfigResource, Config>> describeConfigsFuture = mock(KafkaFuture.class);
        when(describeConfigsFuture.get()).thenThrow(describeConfigsFailure);
        DescribeConfigsResult describeConfigsResult = mock(DescribeConfigsResult.class);
        when(describeConfigsResult.all()).thenReturn(describeConfigsFuture);
        when(sourceAdmin.describeConfigs(any())).thenReturn(describeConfigsResult);

        try (LogCaptureAppender connectorLogs = LogCaptureAppender.createAndRegister(MirrorUtils.class)) {
            connectorLogs.setClassLogger(MirrorUtils.class, Level.TRACE);
            Set<String> topics = new HashSet<>();
            topics.add("topic1");
            topics.add("topic2");
            ExecutionException exception = assertThrows(ExecutionException.class, () -> connector.describeTopicConfigs(topics));
            assertEquals(
                    exception.getCause().getClass().getSimpleName() + " occurred while trying to describe configs for topics [topic1, topic2] on source1 cluster",
                    connectorLogs.getMessages().get(0)
            );
        }
    }