@Nonnull
    @Override
    public Collection<DataConnectionResource> listResources() {
        try (AdminClient client = AdminClient.create(getConfig().getProperties())) {
            return client.listTopics().names().get()
                    .stream()
                    .sorted()
                    .map(n -> new DataConnectionResource("topic", n))
                    .collect(Collectors.toList());
        } catch (ExecutionException | InterruptedException e) {
            throw new HazelcastException("Could not get list of topics for DataConnection " + getConfig().getName(), e);
        }
    }
@Test
    public void list_resources_should_return_empty_list_for_no_topics() {
        kafkaDataConnection = createKafkaDataConnection(kafkaTestSupport);

        Collection<DataConnectionResource> resources = kafkaDataConnection.listResources();
        List<DataConnectionResource> withoutConfluent =
                resources.stream().filter(r -> !Arrays.toString(r.name()).contains("__confluent")).collect(toList());
        assertThat(withoutConfluent).isEmpty();
    }