### Question:
KafkaTopicClientImpl implements KafkaTopicClient { @Override public Set<String> listNonInternalTopicNames() { return listTopicNames().stream() .filter((topic) -> !(topic.startsWith(KsqlConstants.KSQL_INTERNAL_TOPIC_PREFIX) || topic.startsWith(KsqlConstants.CONFLUENT_INTERNAL_TOPIC_PREFIX))) .collect(Collectors.toSet()); } KafkaTopicClientImpl(final AdminClient adminClient); @Override void createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor
  ); @Override void createTopic(
      final String topic,
      final int numPartitions,
      final short replicationFactor,
      final Map<String, ?> configs
  ); @Override boolean isTopicExists(final String topic); @Override Set<String> listTopicNames(); @Override Set<String> listNonInternalTopicNames(); @Override Map<String, TopicDescription> describeTopics(final Collection<String> topicNames); @Override Map<String, String> getTopicConfig(final String topicName); @Override boolean addTopicConfig(final String topicName, final Map<String, ?> overrides); @Override TopicCleanupPolicy getTopicCleanupPolicy(final String topicName); @Override void deleteTopics(final List<String> topicsToDelete); @Override void deleteInternalTopics(final String applicationId); void close();  }### Answer:
@Test public void shouldFilterInternalTopics() { expect(adminClient.listTopics()).andReturn(getListTopicsResultWithInternalTopics()); replay(adminClient); KafkaTopicClient kafkaTopicClient = new KafkaTopicClientImpl(adminClient); Set<String> names = kafkaTopicClient.listNonInternalTopicNames(); assertThat(names, equalTo(Utils.mkSet(topicName1, topicName2, topicName3))); verify(adminClient); }