@Test public void resetTopicConfig_existingConfig_resetsConfig() throws Exception { expect(clusterManager.getCluster(CLUSTER_ID)).andReturn(completedFuture(Optional.of(CLUSTER))); expect( adminClient.describeConfigs( eq(singletonList(new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME))), anyObject(DescribeConfigsOptions.class))) .andReturn(describeConfigsResult); expect(describeConfigsResult.values()) .andReturn( singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME), KafkaFuture.completedFuture(CONFIG))); expect( adminClient.incrementalAlterConfigs( singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME), singletonList( new AlterConfigOp( new ConfigEntry(CONFIG_1.getName(), null), AlterConfigOp.OpType.DELETE))))) .andReturn(alterConfigsResult); expect(alterConfigsResult.values()) .andReturn( singletonMap( new ConfigResource(ConfigResource.Type.TOPIC, TOPIC_NAME), KafkaFuture.completedFuture(null))); replay(clusterManager, adminClient, describeConfigsResult, alterConfigsResult); topicConfigManager.resetTopicConfig( CLUSTER_ID, TOPIC_NAME, CONFIG_1.getName()).get(); verify(adminClient); }
@Override public CompletableFuture<Void> resetTopicConfig( String clusterId, String topicName, String name) { return safeResetConfig( clusterId, new ConfigResource(ConfigResource.Type.TOPIC, topicName), TopicConfig.builder().setClusterId(clusterId).setTopicName(topicName), name); }
TopicConfigManagerImpl extends AbstractConfigManager<TopicConfig, TopicConfig.Builder> implements TopicConfigManager { @Override public CompletableFuture<Void> resetTopicConfig( String clusterId, String topicName, String name) { return safeResetConfig( clusterId, new ConfigResource(ConfigResource.Type.TOPIC, topicName), TopicConfig.builder().setClusterId(clusterId).setTopicName(topicName), name); } }
TopicConfigManagerImpl extends AbstractConfigManager<TopicConfig, TopicConfig.Builder> implements TopicConfigManager { @Override public CompletableFuture<Void> resetTopicConfig( String clusterId, String topicName, String name) { return safeResetConfig( clusterId, new ConfigResource(ConfigResource.Type.TOPIC, topicName), TopicConfig.builder().setClusterId(clusterId).setTopicName(topicName), name); } @Inject  TopicConfigManagerImpl(Admin adminClient, ClusterManager clusterManager); }
TopicConfigManagerImpl extends AbstractConfigManager<TopicConfig, TopicConfig.Builder> implements TopicConfigManager { @Override public CompletableFuture<Void> resetTopicConfig( String clusterId, String topicName, String name) { return safeResetConfig( clusterId, new ConfigResource(ConfigResource.Type.TOPIC, topicName), TopicConfig.builder().setClusterId(clusterId).setTopicName(topicName), name); } @Inject  TopicConfigManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<TopicConfig>> listTopicConfigs(
      String clusterId, String topicName); @Override CompletableFuture<Optional<TopicConfig>> getTopicConfig(
      String clusterId, String topicName, String name); @Override CompletableFuture<Void> updateTopicConfig(
      String clusterId, String topicName, String name, String newValue); @Override CompletableFuture<Void> resetTopicConfig(
      String clusterId, String topicName, String name); @Override CompletableFuture<Void> alterTopicConfigs(
      String clusterId, String topicName, List<AlterConfigCommand> commands); }
TopicConfigManagerImpl extends AbstractConfigManager<TopicConfig, TopicConfig.Builder> implements TopicConfigManager { @Override public CompletableFuture<Void> resetTopicConfig( String clusterId, String topicName, String name) { return safeResetConfig( clusterId, new ConfigResource(ConfigResource.Type.TOPIC, topicName), TopicConfig.builder().setClusterId(clusterId).setTopicName(topicName), name); } @Inject  TopicConfigManagerImpl(Admin adminClient, ClusterManager clusterManager); @Override CompletableFuture<List<TopicConfig>> listTopicConfigs(
      String clusterId, String topicName); @Override CompletableFuture<Optional<TopicConfig>> getTopicConfig(
      String clusterId, String topicName, String name); @Override CompletableFuture<Void> updateTopicConfig(
      String clusterId, String topicName, String name, String newValue); @Override CompletableFuture<Void> resetTopicConfig(
      String clusterId, String topicName, String name); @Override CompletableFuture<Void> alterTopicConfigs(
      String clusterId, String topicName, List<AlterConfigCommand> commands);  }