@Override
  public boolean addTopicConfig(final String topicName, final Map<String, ?> overrides) {
    final ConfigResource resource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);

    final Map<String, String> stringConfigs = toStringConfigs(overrides);

    try {
      final Map<String, String> existingConfig = topicConfig(topicName, false);

      final boolean changed = stringConfigs.entrySet().stream()
          .anyMatch(e -> !Objects.equals(existingConfig.get(e.getKey()), e.getValue()));
      if (!changed) {
        return false;
      }

      final Set<AlterConfigOp> entries = stringConfigs.entrySet().stream()
          .map(e -> new ConfigEntry(e.getKey(), e.getValue()))
          .map(ce -> new AlterConfigOp(ce, AlterConfigOp.OpType.SET))
          .collect(Collectors.toSet());

      final Map<ConfigResource, Collection<AlterConfigOp>> request =
          Collections.singletonMap(resource, entries);

      ExecutorUtil.executeWithRetries(
          () -> adminClient.get().incrementalAlterConfigs(request).all().get(),
          ExecutorUtil.RetryBehaviour.ON_RETRYABLE);

      return true;
    } catch (final UnsupportedVersionException e) {
      return addTopicConfigLegacy(topicName, stringConfigs);
    } catch (final Exception e) {
      throw new KafkaResponseGetFailedException(
          "Failed to set config for Kafka Topic " + topicName, e);
    }
  }
@Test
  public void shouldRetryAddingTopicConfig() {
    // Given:
    givenTopicConfigs(
        "peter",
        overriddenConfigEntry(TopicConfig.RETENTION_MS_CONFIG, "12345"),
        defaultConfigEntry(TopicConfig.COMPRESSION_TYPE_CONFIG, "snappy")
    );

    final Map<String, ?> overrides = ImmutableMap.of(
        CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT
    );

    when(adminClient.incrementalAlterConfigs(any()))
        .thenAnswer(alterConfigsResult(new DisconnectException()))
        .thenAnswer(alterConfigsResult());

    // When:
    kafkaTopicClient.addTopicConfig("peter", overrides);

    // Then:
    verify(adminClient, times(2)).incrementalAlterConfigs(any());
  }