fix: Implement new KIP-455 AdminClient AlterPartitionReassignments an…

…d tPartitionReassignments APIs (#3218)
confluentinc · Aug 15, 2019 · d951026 · d951026
1 parent c860793
commit d951026
Show file tree

Hide file tree

Showing 2 changed files with 22 additions and 0 deletions.
diff --git a/ksql-engine/src/main/java/io/confluent/ksql/services/SandboxedAdminClient.java b/ksql-engine/src/main/java/io/confluent/ksql/services/SandboxedAdminClient.java
@@ -18,11 +18,14 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Map;
+import java.util.Optional;
 import java.util.Set;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.AlterConfigsOptions;
 import org.apache.kafka.clients.admin.AlterConfigsResult;
+import org.apache.kafka.clients.admin.AlterPartitionReassignmentsOptions;
+import org.apache.kafka.clients.admin.AlterPartitionReassignmentsResult;
 import org.apache.kafka.clients.admin.AlterReplicaLogDirsOptions;
 import org.apache.kafka.clients.admin.AlterReplicaLogDirsResult;
 import org.apache.kafka.clients.admin.Config;
@@ -66,8 +69,11 @@
 import org.apache.kafka.clients.admin.ListConsumerGroupOffsetsResult;
 import org.apache.kafka.clients.admin.ListConsumerGroupsOptions;
 import org.apache.kafka.clients.admin.ListConsumerGroupsResult;
+import org.apache.kafka.clients.admin.ListPartitionReassignmentsOptions;
+import org.apache.kafka.clients.admin.ListPartitionReassignmentsResult;
 import org.apache.kafka.clients.admin.ListTopicsOptions;
 import org.apache.kafka.clients.admin.ListTopicsResult;
+import org.apache.kafka.clients.admin.NewPartitionReassignment;
 import org.apache.kafka.clients.admin.NewPartitions;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.apache.kafka.clients.admin.RecordsToDelete;
@@ -299,6 +305,20 @@ public ElectLeadersResult electLeaders(
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public AlterPartitionReassignmentsResult alterPartitionReassignments(
+      final Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments,
+      final AlterPartitionReassignmentsOptions options) {
+    throw new UnsupportedOperationException();
+  }
+
+  @Override
+  public ListPartitionReassignmentsResult listPartitionReassignments(
+      final Optional<Set<TopicPartition>> partitions,
+      final ListPartitionReassignmentsOptions options) {
+    throw new UnsupportedOperationException();
+  }
+
   @Override
   public Map<MetricName, ? extends Metric> metrics() {
     throw new UnsupportedOperationException();

diff --git a/ksql-engine/src/test/java/io/confluent/ksql/services/SandboxedAdminClientTest.java b/ksql-engine/src/test/java/io/confluent/ksql/services/SandboxedAdminClientTest.java
@@ -20,6 +20,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.TimeUnit;
 import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClient;
@@ -46,6 +47,7 @@ public static Collection<TestCase<Admin>> getMethodsToTest() {
           .ignore("close", Duration.class)
           .ignore("close", long.class, TimeUnit.class)
           .setDefault(ElectLeadersOptions.class, new ElectLeadersOptions())
+          .setDefault(Optional.class, Optional.empty())
           .build();
     }