Add a pagination thread configuration to allow parallel resolution of…

… list page
tchiotludo · Nov 11, 2019 · 8fd76c1 · 8fd76c1
1 parent dad1fa2
commit 8fd76c1
Show file tree

Hide file tree

Showing 11 changed files with 84 additions and 70 deletions.
diff --git a/README.md b/README.md
@@ -153,8 +153,11 @@ kafkahq:
 
 ### KafkaHQ configuration 
 
+#### Pagination
+* `kafkahq.pagination.page-size` number of topics per page (default : 25)
+* `kafkahq.pagination.threads` number of parallel threads to resolve current page (default : 16). This setting can have a signifiant impact on performance on list page since it will fetch in parallel the Kafka API.
+
 #### Topic List 
-* `kafkahq.topic.page-size` number of topics per page (default : 25)
 * `kafkahq.topic.default-view` is default list view (ALL, HIDE_INTERNAL, HIDE_INTERNAL_STREAM, HIDE_STREAM)
 * `kafkahq.topic.internal-regexps` is list of regexp to be considered as internal (internal topic can't be deleted or updated)
 * `kafkahq.topic.stream-regexps` is list of regexp to be considered as internal stream topic
@@ -173,13 +176,6 @@ These parameters are the default values used in the topic creation page.
 * `kafkahq.topic-data.poll-timeout`: The time, in milliseconds, spent waiting in poll if data is not available in the
   buffer (default: 1000).
 
-#### Consumer Groups List 
-* `kafkahq.consumer-groups.page-size` number of consumer groups per page (default : 25)
-
-
-#### Schema List 
-* `kafkahq.schema.page-size` number of schemas per page (default : 25)
-
 
 ### Security
 * `kafkahq.security.default-roles`: Roles available for all the user even unlogged user, roles available are :

diff --git a/application.example.yml b/application.example.yml
@@ -57,12 +57,15 @@ kafkahq:
         ssl.keystore.password: password
         ssl.key.password: password
 
+  pagination:
+    page-size: 25 # number of elements per page (default : 25)
+    threads: 16 # Number of parallel threads to resolve page
+
   # Topic list display options (optional)
   topic:
     retention: 172800000 # default retention period when creating topic
     partition: 3 #  default number of partition when creating topic
     replication: 3 # default number of replicas when creating topic
-    page-size: 25 # number of topics per page (default : 25)
     default-view: HIDE_INTERNAL # default list view (ALL, HIDE_INTERNAL, HIDE_INTERNAL_STREAM, HIDE_STREAM)
     internal-regexps: # list of regexp to be considered as internal (internal topic can't be deleted or updated)
       - "^_.*$"
@@ -81,14 +84,6 @@ kafkahq:
     size: 50 # max record per page (default: 50)
     poll-timeout: 1000 # The time, in milliseconds, spent waiting in poll if data is not available in the buffer.
 
-  # Schema list display options (optional)
-  schema:
-    page-size: 25 # number of schemas per page (default : 25)
-
-  # Consumer groups list display options (optional)
-  consumer-groups:
-    page-size: 25 # number of consumer groups per page (default : 25)
-
   # Auth & Roles (optional)
   security:
     default-roles: # Roles available for all the user even unlogged user

diff --git a/docker-compose-dev.yml b/docker-compose-dev.yml
@@ -61,6 +61,8 @@ services:
       KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: 'false'
       KAFKA_JMX_PORT: 9091
       KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
+      KAFKA_AUTHORIZER_CLASS_NAME: 'kafka.security.auth.SimpleAclAuthorizer'
+      KAFKA_ALLOW_EVERYONE_IF_NO_ACL_FOUND: 'true'
     links:
       - zookeeper
     ports:

diff --git a/package-lock.json b/package-lock.json
diff --git a/src/main/java/org/kafkahq/controllers/GroupController.java b/src/main/java/org/kafkahq/controllers/GroupController.java
@@ -19,6 +19,7 @@
 import org.kafkahq.repositories.ConsumerGroupRepository;
 import org.kafkahq.repositories.RecordRepository;
 import org.kafkahq.utils.CompletablePaged;
+import org.kafkahq.utils.CompletablePagedService;
 
 import javax.inject.Inject;
 import java.time.Instant;
@@ -35,14 +36,17 @@
 public class GroupController extends AbstractController {
     private ConsumerGroupRepository consumerGroupRepository;
     private RecordRepository recordRepository;
-
-    @Value("${kafkahq.consumer-groups.page-size:25}")
-    private Integer pageSize;
+    private CompletablePagedService completablePagedService;
 
     @Inject
-    public GroupController(ConsumerGroupRepository consumerGroupRepository, RecordRepository recordRepository) {
+    public GroupController(
+        ConsumerGroupRepository consumerGroupRepository,
+        RecordRepository recordRepository,
+        CompletablePagedService completablePagedService
+    ) {
         this.consumerGroupRepository = consumerGroupRepository;
         this.recordRepository = recordRepository;
+        this.completablePagedService = completablePagedService;
     }
 
     @View("groupList")
@@ -51,13 +55,7 @@ public HttpResponse list(HttpRequest request, String cluster, Optional<String> s
 
         List<CompletableFuture<ConsumerGroup>> list = this.consumerGroupRepository.list(cluster, search);
         URIBuilder uri = URIBuilder.fromURI(request.getUri());
-
-        CompletablePaged<ConsumerGroup> paged = new CompletablePaged<>(
-            list,
-            this.pageSize,
-            uri,
-            page.orElse(1)
-        );
+        CompletablePaged<ConsumerGroup> paged = completablePagedService.of(list, uri, page.orElse(1));
 
         return this.template(
             request,

diff --git a/src/main/java/org/kafkahq/controllers/SchemaController.java b/src/main/java/org/kafkahq/controllers/SchemaController.java
@@ -2,7 +2,6 @@
 
 import com.google.common.collect.ImmutableMap;
 import io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException;
-import io.micronaut.context.annotation.Value;
 import io.micronaut.http.*;
 import io.micronaut.http.annotation.Controller;
 import io.micronaut.http.annotation.Get;
@@ -15,6 +14,7 @@
 import org.kafkahq.modules.RequestHelper;
 import org.kafkahq.repositories.SchemaRegistryRepository;
 import org.kafkahq.utils.CompletablePaged;
+import org.kafkahq.utils.CompletablePagedService;
 
 import javax.inject.Inject;
 import java.io.IOException;
@@ -29,13 +29,12 @@
 @Controller("${kafkahq.server.base-path:}/{cluster}/schema")
 public class SchemaController extends AbstractController {
     private SchemaRegistryRepository schemaRepository;
-
-    @Value("${kafkahq.schema.page-size:25}")
-    private Integer pageSize;
+    private CompletablePagedService completablePagedService;
 
     @Inject
-    public SchemaController(SchemaRegistryRepository schemaRepository) {
+    public SchemaController(SchemaRegistryRepository schemaRepository, CompletablePagedService completablePagedService) {
         this.schemaRepository = schemaRepository;
+        this.completablePagedService = completablePagedService;
     }
 
     @View("schemaList")
@@ -49,12 +48,7 @@ public HttpResponse list(
         List<CompletableFuture<Schema>> list = this.schemaRepository.getAll(cluster, search);
 
         URIBuilder uri = URIBuilder.fromURI(request.getUri());
-        CompletablePaged<Schema> paged = new CompletablePaged<>(
-            list,
-            this.pageSize,
-            uri,
-            page.orElse(1)
-        );
+        CompletablePaged<Schema> paged = completablePagedService.of(list, uri, page.orElse(1));
 
         return this.template(
             request,

diff --git a/src/main/java/org/kafkahq/controllers/TopicController.java b/src/main/java/org/kafkahq/controllers/TopicController.java
@@ -28,6 +28,7 @@
 import org.kafkahq.repositories.RecordRepository;
 import org.kafkahq.repositories.TopicRepository;
 import org.kafkahq.utils.CompletablePaged;
+import org.kafkahq.utils.CompletablePagedService;
 import org.reactivestreams.Publisher;
 
 import javax.inject.Inject;
@@ -47,10 +48,10 @@ public class TopicController extends AbstractController {
     private RecordRepository recordRepository;
     private FreemarkerViewsRenderer freemarkerViewsRenderer;
     private Environment environment;
+    private CompletablePagedService completablePagedService;
+
     @Value("${kafkahq.topic.default-view}")
     private String defaultView;
-    @Value("${kafkahq.topic.page-size:25}")
-    private Integer pageSize;
     @Value("${kafkahq.topic.replication}")
     private Integer replicationFactor;
     @Value("${kafkahq.topic.retention}")
@@ -59,17 +60,20 @@ public class TopicController extends AbstractController {
     private Integer partitionCount;
 
     @Inject
-    public TopicController(TopicRepository topicRepository,
-                           ConfigRepository configRepository,
-                           RecordRepository recordRepository,
-                           FreemarkerViewsRenderer freemarkerViewsRenderer,
-                           Environment environment)
-    {
+    public TopicController(
+        TopicRepository topicRepository,
+        ConfigRepository configRepository,
+        RecordRepository recordRepository,
+        FreemarkerViewsRenderer freemarkerViewsRenderer,
+        Environment environment,
+        CompletablePagedService completablePagedService
+    ) {
         this.topicRepository = topicRepository;
         this.configRepository = configRepository;
         this.recordRepository = recordRepository;
         this.freemarkerViewsRenderer = freemarkerViewsRenderer;
         this.environment = environment;
+        this.completablePagedService = completablePagedService;
     }
 
     @View("topicList")
@@ -88,12 +92,7 @@ public HttpResponse list(
         );
 
         URIBuilder uri = URIBuilder.fromURI(request.getUri());
-        CompletablePaged<Topic> paged = new CompletablePaged<>(
-            list,
-            this.pageSize,
-            uri,
-            page.orElse(1)
-        );
+        CompletablePaged<Topic> paged = completablePagedService.of(list, uri, page.orElse(1));
 
         return this.template(
             request,

diff --git a/src/main/java/org/kafkahq/utils/CompletablePaged.java b/src/main/java/org/kafkahq/utils/CompletablePaged.java
@@ -4,14 +4,14 @@
 import org.codehaus.httpcache4j.uri.URIBuilder;
 
 import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
+import java.util.concurrent.*;
 import java.util.stream.Collectors;
 
 @AllArgsConstructor
 public class CompletablePaged <T> {
     private List<CompletableFuture<T>> list;
     private Integer pageSize;
+    private ExecutorService executorService;
     private URIBuilder uri;
     private Integer currentPage;
 
@@ -20,6 +20,7 @@ public int size() {
     }
 
     public URIBuilder before() {
+
         if (currentPage - 1 > 0) {
             return uri.addParameter("page", String.valueOf(currentPage - 1));
         } else {
@@ -67,11 +68,13 @@ public List<T> complete() throws ExecutionException, InterruptedException {
             futuresList.toArray(new CompletableFuture[0])
         );
 
-        return allFuturesResult.thenApply(s ->
+        return allFuturesResult.thenApplyAsync(s ->
             futuresList.stream().
                 map(CompletableFuture::join).
-                collect(Collectors.toList())
+                collect(Collectors.toList()),
+            this.executorService
         )
             .get();
+
     }
 }
diff --git a/src/main/java/org/kafkahq/utils/CompletablePagedService.java b/src/main/java/org/kafkahq/utils/CompletablePagedService.java
@@ -0,0 +1,30 @@
+package org.kafkahq.utils;
+
+import io.micronaut.context.annotation.Value;
+import org.codehaus.httpcache4j.uri.URIBuilder;
+
+import javax.inject.Singleton;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Singleton
+public class CompletablePagedService {
+    private Integer pageSize;
+    private Integer threads;
+    private ExecutorService executorService;
+
+    public CompletablePagedService(
+        @Value("${kafkahq.pagination.page-size}") Integer pageSize,
+        @Value("${kafkahq.pagination.threads}") Integer threads
+    ) {
+        this.pageSize = pageSize;
+        this.threads = threads;
+        executorService = Executors.newFixedThreadPool(this.threads);
+    }
+
+    public <T> CompletablePaged<T> of(List<CompletableFuture<T>> list, URIBuilder uri, Integer currentPage) {
+        return new CompletablePaged<>(list, this.pageSize, executorService, uri, currentPage);
+    }
+}
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
@@ -80,9 +80,12 @@ kafkahq:
         enable.auto.commit: "false"
         default.api.timeout.ms: 15000
 
+  pagination:
+    page-size: 25
+    threads: 16
+
   topic:
     default-view: HIDE_INTERNAL
-    page-size: 25
     replication: 1
     retention: 86400000
     partition: 1
@@ -97,12 +100,6 @@ kafkahq:
       - "^.*-repartition$"
       - "^.*-rekey$"
 
-  schema:
-    page-size: 25
-
-  consumer-groups:
-    page-size: 25
-
   topic-data:
     sort: OLDEST
     size: 50

diff --git a/src/main/resources/views/topicList.ftl b/src/main/resources/views/topicList.ftl
@@ -63,7 +63,7 @@
                             </span>
                         </td>
                         <td>
-                            <#if topic.getLogDirSize()?? || topic.getLogDirSize().isEmpty() >
+                            <#if topic.getLogDirSize().isEmpty() >
                                 n/a
                             <#else>
                                 ${functions.filesize(topic.getLogDirSize().get())}
-Original file line number
+Diff line change
@@ Expand Up / @@ -63,7 +63,7 @@ @@
                                 </span>
                             </td>
                             <td>
-                                <#if topic.getLogDirSize()?? || topic.getLogDirSize().isEmpty() >
+                                <#if topic.getLogDirSize().isEmpty() >
                                     n/a
                                 <#else>
                                     ${functions.filesize(topic.getLogDirSize().get())}
@@ Expand Down @@