@Override
  public TopicCleanupPolicy getTopicCleanupPolicy(final String topicName) {
    final String policy = getTopicConfig(topicName)
        .getOrDefault(TopicConfig.CLEANUP_POLICY_CONFIG, "")
        .toLowerCase();

    if (policy.equals("compact")) {
      return TopicCleanupPolicy.COMPACT;
    } else if (policy.equals("delete")) {
      return TopicCleanupPolicy.DELETE;
    } else if (policy.contains("compact") && policy.contains("delete")) {
      return TopicCleanupPolicy.COMPACT_DELETE;
    } else {
      throw new KsqlException("Could not get the topic configs for : " + topicName);
    }
  }
@Test
  public void shouldGetTopicCleanUpPolicyCompact() {
    // Given:
    givenTopicConfigs(
        "foo",
        overriddenConfigEntry(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT)
    );

    // When / Then:
    assertThat(kafkaTopicClient.getTopicCleanupPolicy("foo"),
        is(TopicCleanupPolicy.COMPACT));
  }