Use blank ID during Kafka creation, only fetch leader's storage, test

Signed-off-by: Michael Edgar <[email protected]>

Author: MikeEdgar

api/src/main/java/com/github/eyefloaters/console/api/model/KafkaCluster.java

@@ -177,7 +177,7 @@ public static KafkaCluster fromCursor(JsonObject cursor) {
 
     public String toCursor(List<String> sortFields) {
         JsonObjectBuilder cursor = Json.createObjectBuilder()
-                .add("id", id == null ? JsonValue.NULL : Json.createValue(id));
+                .add("id", id == null ? Json.createValue("") : Json.createValue(id));
 
         JsonObjectBuilder attrBuilder = Json.createObjectBuilder();
         maybeAddAttribute(attrBuilder, sortFields, Fields.NAME, name);

api/src/main/java/com/github/eyefloaters/console/api/service/TopicService.java

@@ -50,7 +50,6 @@
 import com.github.eyefloaters.console.api.model.OffsetInfo;
 import com.github.eyefloaters.console.api.model.PartitionId;
 import com.github.eyefloaters.console.api.model.PartitionInfo;
-import com.github.eyefloaters.console.api.model.PartitionReplica;
 import com.github.eyefloaters.console.api.model.ReplicaLocalStorage;
 import com.github.eyefloaters.console.api.model.Topic;
 import com.github.eyefloaters.console.api.model.TopicPatch;
@@ -486,7 +485,7 @@ CompletionStage<Void> listOffsets(Admin adminClient, Map<Uuid, Either<Topic, Thr
 
         var pendingOffsets = getRequestOffsetSpecs(offsetSpec)
             .stream()
-            .map(reqOffsetSpec -> topicPartitionReplicas(onlineTopics, topicIds)
+            .map(reqOffsetSpec -> topicPartitionLeaders(onlineTopics, topicIds)
                 .keySet()
                 .stream()
                 .collect(Collectors.toMap(Function.identity(), ignored -> reqOffsetSpec)))
@@ -513,20 +512,19 @@ List<OffsetSpec> getRequestOffsetSpecs(String offsetSpec) {
     }
 
     /**
-     * Build of map of {@linkplain PartitionId}s to the list of replicas where
-     * the partitions are placed. Concurrently, a map of topic names to topic
-     * identifiers is constructed to support cross referencing the
-     * {@linkplain PartitionId} keys (via {@linkplain PartitionId#topicId()})
-     * back to the topic's {@linkplain Uuid}. This allows easy access of the topics
-     * located in the topics map provided to this method and is particularly useful
-     * for Kafka operations that still require topic name.
+     * Build of map of {@linkplain PartitionId}s to the partition leader node ID.
+     * Concurrently, a map of topic names to topic identifiers is constructed to
+     * support cross referencing the {@linkplain PartitionId} keys (via
+     * {@linkplain PartitionId#topicId()}) back to the topic's {@linkplain Uuid}.
+     * This allows easy access of the topics located in the topics map provided to
+     * this method and is particularly useful for Kafka operations that still
+     * require topic name.
      *
      * @param topics   map of topics (keyed by Id)
      * @param topicIds map of topic names to topic Ids, modified by this method
-     * @return map of {@linkplain PartitionId}s to the list of replicas where the
-     *         partitions are placed
+     * @return map of {@linkplain PartitionId}s to the partition leader node ID
      */
-    Map<PartitionId, List<Integer>> topicPartitionReplicas(Map<Uuid, Either<Topic, Throwable>> topics, Map<String, Uuid> topicIds) {
+    Map<PartitionId, Integer> topicPartitionLeaders(Map<Uuid, Either<Topic, Throwable>> topics, Map<String, Uuid> topicIds) {
         return topics.entrySet()
                 .stream()
                 .filter(entry -> entry.getValue().isPrimaryPresent())
@@ -541,8 +539,7 @@ Map<PartitionId, List<Integer>> topicPartitionReplicas(Map<Uuid, Either<Topic, T
                         .filter(PartitionInfo::online)
                         .map(partition -> {
                             var key = new PartitionId(topic.getId(), topic.name(), partition.getPartition());
-                            List<Integer> value = partition.getReplicas().stream().map(PartitionReplica::nodeId).toList();
-                            return Map.entry(key, value);
+                            return Map.entry(key, partition.getLeaderId());
                         }))
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
     }
@@ -609,15 +606,14 @@ Either<OffsetInfo, Throwable> either(ListOffsetsResultInfo result, Throwable err
     CompletionStage<Void> describeLogDirs(Admin adminClient, Map<Uuid, Either<Topic, Throwable>> topics) {
         Map<String, Uuid> topicIds = new HashMap<>(topics.size());
 
-        var topicPartitionReplicas = topicPartitionReplicas(topics, topicIds);
-        var nodeIds = topicPartitionReplicas.values().stream().flatMap(Collection::stream).distinct().toList();
+        var topicPartitionReplicas = topicPartitionLeaders(topics, topicIds);
+        var nodeIds = topicPartitionReplicas.values().stream().distinct().toList();
         var logDirs = adminClient.describeLogDirs(nodeIds, new DescribeLogDirsOptions()
                 .timeoutMs(5000))
                 .descriptions();
 
         var pendingInfo = topicPartitionReplicas.entrySet()
             .stream()
-            .flatMap(e -> e.getValue().stream().map(node -> Map.entry(e.getKey(), node)))
             .map(e -> {
                 var topicPartition = e.getKey().toKafkaModel();
                 int nodeId = e.getValue();

api/src/test/java/com/github/eyefloaters/console/api/TopicsResourceIT.java

@@ -41,9 +41,13 @@
 import org.apache.kafka.clients.admin.ListOffsetsResult.ListOffsetsResultInfo;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicCollection;
 import org.apache.kafka.common.TopicCollection.TopicIdCollection;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.acl.AclOperation;
 import org.apache.kafka.common.config.ConfigResource;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.internals.KafkaFutureImpl;
@@ -125,28 +129,29 @@ class TopicsResourceIT {
     TopicHelper topicUtils;
     ConsumerUtils groupUtils;
     String clusterId1;
+    URI bootstrapServers1;
     String clusterId2;
     ServerSocket randomSocket;
 
     @BeforeEach
     void setup() throws IOException {
-        URI bootstrapServers = URI.create(deployments.getExternalBootstrapServers());
+        bootstrapServers1 = URI.create(deployments.getExternalBootstrapServers());
         randomSocket = new ServerSocket(0);
         URI randomBootstrapServers = URI.create("dummy://localhost:" + randomSocket.getLocalPort());
 
-        topicUtils = new TopicHelper(bootstrapServers, config, null);
+        topicUtils = new TopicHelper(bootstrapServers1, config, null);
         topicUtils.deleteAllTopics();
 
         groupUtils = new ConsumerUtils(config, null);
 
-        utils = new TestHelper(bootstrapServers, config, null);
+        utils = new TestHelper(bootstrapServers1, config, null);
 
         clusterId1 = utils.getClusterId();
         clusterId2 = UUID.randomUUID().toString();
 
         client.resources(Kafka.class).delete();
         client.resources(Kafka.class)
-            .resource(utils.buildKafkaResource("test-kafka1", clusterId1, bootstrapServers))
+            .resource(utils.buildKafkaResource("test-kafka1", clusterId1, bootstrapServers1))
             .create();
         // Second cluster is offline/non-existent
         client.resources(Kafka.class)
@@ -955,6 +960,60 @@ void testDescribeTopicWithNoSuchTopic() {
             .body("errors.code", contains("4041"));
     }
 
+    @Test
+    void testDescribeTopicWithOfflinePartition() {
+        String topicName = UUID.randomUUID().toString();
+        Map<String, String> topicIds = topicUtils.createTopics(clusterId1, List.of(topicName), 2);
+
+        //int partition, Node leader, List<Node> replicas, List<Node> isr
+        Node node0 = new Node(0, "node0", bootstrapServers1.getPort());
+        Node node1 = new Node(1, "node1", bootstrapServers1.getPort());
+
+        Answer<DescribeTopicsResult> describeTopicsResult = args -> {
+            List<TopicPartitionInfo> partitions = List.of(
+                    // Online, 2 replicas, 1 ISR
+                    new TopicPartitionInfo(0, node0, List.of(node0, node1), List.of(node0)),
+                    // Offline, 2 replicas, no ISRs
+                    new TopicPartitionInfo(1, null, List.of(node0, node1), List.of()));
+            Set<AclOperation> authorizedOperations = Set.of(AclOperation.ALL);
+            Uuid topicId = Uuid.fromString(topicIds.get(topicName));
+
+            var description = KafkaFuture.completedFuture(
+                    new TopicDescription(topicName, false, partitions, authorizedOperations, topicId));
+
+            class Result extends DescribeTopicsResult {
+                Result() {
+                    super(Map.of(topicId, description), null);
+                }
+            }
+
+            return new Result();
+        };
+
+        AdminClientSpy.install(client -> {
+            // Mock listOffsets
+            doAnswer(describeTopicsResult)
+                .when(client)
+                .describeTopics(any(TopicCollection.class), any(DescribeTopicsOptions.class));
+        });
+
+        whenRequesting(req -> req.get("{topicId}", clusterId1, topicIds.get(topicName)))
+            .assertThat()
+            .statusCode(is(Status.OK.getStatusCode()))
+            .body("data.attributes.name", is(topicName))
+            .body("data.attributes.status", is("PartiallyOffline"))
+            .body("data.attributes.partitions", hasSize(2))
+            .body("data.attributes.partitions[0].status", is("UnderReplicated"))
+            .body("data.attributes.partitions[0].replicas[0].inSync", is(true))
+            .body("data.attributes.partitions[0].replicas[0].localStorage", notNullValue())
+            // storage not fetched for followers
+            .body("data.attributes.partitions[0].replicas[1].inSync", is(false))
+            .body("data.attributes.partitions[0].replicas[1].localStorage", nullValue())
+            // Partition 2, offline with no ISRs
+            .body("data.attributes.partitions[1].status", is("Offline"))
+            .body("data.attributes.partitions[1].replicas.inSync", everyItem(is(false)));
+    }
+
     @Test
     void testCreateTopicSucceeds() {
         String topicName = UUID.randomUUID().toString();