Monitoring consumer groups #1650

WojciechSova · 2023-01-18T09:26:07Z

No description provided.

faderskd · 2023-02-17T14:21:13Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+
+    private Optional<ConsumerGroupDescription> getConsumerGroupDescription(ConsumerGroupId consumerGroupId)
+            throws InterruptedException, ExecutionException {
+        Optional<ConsumerGroupDescription> consumerGroupDescription = adminClient


return adminClient.describe...`, no need to declare variable

faderskd · 2023-02-17T14:22:30Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+    }
+
+    private static int getPartitionsInConsumerGroups(Optional<ConsumerGroupDescription> consumerGroupDescription) {
+        return consumerGroupDescription.get().members().stream()


why do you use optional if you don't check if it is present ?

faderskd · 2023-02-17T14:22:47Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+                .mapToInt(member -> member.assignment().topicPartitions().size()).sum();
+    }
+
+    private boolean isConsumerGroupRebalancing(Optional<ConsumerGroupDescription> description) {


same as above

faderskd · 2023-03-09T14:23:40Z

hermes-api/src/main/java/pl/allegro/tech/hermes/api/TopicAndSubscription.java

+    }
+
+    @JsonIgnore
+    public Topic getTopic() {


do you need this method ? it is not used anywhere

faderskd · 2023-03-09T15:04:23Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+    private void monitorSubscriptionsPartitions() {
+        List<List<TopicAndSubscription>> splitSubscriptions = getSplitActiveSubscriptions();
+
+        ExecutorService executorService = Executors.newFixedThreadPool(monitoringProperties.getNumberOfThreads());


creating threads is quite lengthy process, you can create the executor service only once in a constructor

What about monitoring that has not finished in time? A new start of monitoring will cause errors as not all threads are available.

Are you sure? If I understand correctly monitorSubscriptionsPartitions is always executed on scheduledExecutorService which is a singleThreadedExecutor. The scheduledExecutorService thread is blocked until all monitoring tasks finish. Another run won't start if the previous hasn't finished, from SchedulerExecutorService#scheduleAtFixedRate:

If any execution of this task takes longer than its period, then subsequent executions may start late, but will not concurrently execute.

faderskd · 2023-03-09T15:42:26Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+    }
+
+    private List<TopicAndSubscription> createTopicSubscriptions(Topic topic, List<String> subscriptions) {
+        List<TopicAndSubscription> topicAndSubscriptions = new ArrayList<>();


[minor] you can use shorter version too :

return subscriptions.stream().map(it -> new TopicAndSubscription(topic, it)).collect(toList());

faderskd · 2023-03-09T15:56:27Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+                return new ArrayList<TopicAndSubscription>();
+            } finally {
+                logger.info("Monitoring ended");
+                executorService.shutdown();


you can use @PreDestroy annotation to shutdown the executor when will be moved to the class field

faderskd · 2023-03-09T16:13:39Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+
+        List<Future<List<TopicAndSubscription>>> futures = splitSubscriptions.stream().map(it -> executorService.submit(() -> {
+            try {
+                List<MonitoringService> monitoringServices = createMonitoringService();


createMonitoringServices()

faderskd · 2023-03-09T16:32:21Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+    }
+
+    private static void waitForAllThreadsToEnd(List<Future<List<TopicAndSubscription>>> futures) {
+        while (futures.stream().anyMatch(future -> !future.isDone())) {


The easiest way to wait for all futures is

for (Future<?> f : futures) { f.get(); }

faderskd · 2023-03-09T16:44:08Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+            try {
+                return future.get();
+            } catch (InterruptedException | ExecutionException e) {
+                throw new RuntimeException(e);


Throw some more domain specific, custom exception. Just like MonitoringCacheException"Error during monitoring subscriptions...", e)

faderskd · 2023-03-09T17:01:54Z

What about testing ? We should have at least integration test for the happy path. I think we can skip the test when there is not enough consumer groups because it can be hard to reproduce.

# Conflicts: # hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/ManagementConfiguration.java

moscicky · 2023-07-31T13:43:00Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+    }
+
+    private List<List<TopicAndSubscription>> splitSubscriptions(List<TopicAndSubscription> topicAndSubscriptions) {
+        return Lists.partition(topicAndSubscriptions, (topicAndSubscriptions.size() / monitoringProperties.getNumberOfThreads()) + 1);


question: why +1?

From documentation:
Returns consecutive sublists of a list, each of the same size (the final list may be smaller). For example, partitioning a list containing [a, b, c, d, e] with a partition size of 3 yields [[a, b, c], [d, e]] -- an outer list containing two inner lists of three and two elements, all in the original order.
For example:
We have 7 subscriptions and 6 threads.
So the subscription.size() / numberOfThreads = 7/6 = 1. Keep in mind that the subscriptions and the numberOfThreads are integers.
Now the Lists.partition(subscription, 7/6=1) will give us 7 partitions and we only have 6 threads.
Now you can ask that +1 will result in this example in only 4 partitions, but we have more than 7 subscriptions :D

moscicky · 2023-07-31T13:44:19Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+        return splitSubscriptions(getAllActiveSubscriptions());
+    }
+
+    private List<List<TopicAndSubscription>> splitSubscriptions(List<TopicAndSubscription> topicAndSubscriptions) {


nit: why not partitionSubscriptions? Then we could also rename getSplitActiveSubscriptions to getPartitionedActiveSubscriptions which imo sound better

moscicky · 2023-08-08T14:11:08Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+    private void monitorSubscriptionsPartitions() {
+        List<List<TopicAndSubscription>> splitSubscriptions = getSplitActiveSubscriptions();
+
+        ExecutorService executorService = Executors.newFixedThreadPool(monitoringProperties.getNumberOfThreads());


Are you sure? If I understand correctly monitorSubscriptionsPartitions is always executed on scheduledExecutorService which is a singleThreadedExecutor. The scheduledExecutorService thread is blocked until all monitoring tasks finish. Another run won't start if the previous hasn't finished, from SchedulerExecutorService#scheduleAtFixedRate:

If any execution of this task takes longer than its period, then subsequent executions may start late, but will not concurrently execute.

moscicky · 2023-08-08T14:12:17Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+        updateSubscriptionsWithUnassignedPartitionsCache(subscriptionsWithUnassignedPartitions);
+    }
+
+    private List<TopicAndSubscription> getFreshListOfSubscriptionsWithUnassignedPartitions(


nit: really verbose method name, perhaps we can come up with sth simpler?

moscicky · 2023-08-08T14:17:13Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+                List<List<TopicAndSubscription>> splitSubscriptions) {
+        ExecutorService executorService = Executors.newFixedThreadPool(monitoringProperties.getNumberOfThreads());
+
+        List<Future<List<TopicAndSubscription>>> futures = splitSubscriptions.stream().map(it -> executorService.submit(() -> {


suggestion: that's a huge lambda (task) and its purpose is not clear (we only know it produces futures), i'd suggest we move it to separate method

moscicky · 2023-08-08T14:19:44Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+        return Lists.partition(topicAndSubscriptions, (topicAndSubscriptions.size() / monitoringProperties.getNumberOfThreads()) + 1);
+    }
+
+    private List<TopicAndSubscription> checkIfAllPartitionsAreAssignedToSubscriptions(List<TopicAndSubscription> subscriptions,


nit: check usually suggest boolean condition test or sth similar. Perhaps we we should use find or sth similar

moscicky · 2023-08-08T14:28:44Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+            return checkIfAllPartitionsAreAssigned(consumerGroupId, kafkaTopics);
+        } catch (Exception e) {
+            logger.error("Failed to describe group with id: {}", consumerGroupId.asString(), e);
+            return true;


so in case of any error this will fail silently and return that the subscription is valid. imo errors can suggest that sth is wrong with the subscription and the result should be false. Or perhaps this method should have 3 possible results, true/false/error

moscicky · 2023-08-08T14:31:29Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+
+        if (consumerGroupDescription.isEmpty()) {
+            logger.info("Monitoring. Cannot get consumer group description about: {}", consumerGroupId.asString());
+            return true;


again, maybe there should be a third - exceptional outcome?

moscicky · 2023-08-08T14:36:07Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+        }
+
+        int topicPartitions = getTopicPartitions(kafkaTopics);
+        int partitionsInConsumerGroups = getPartitionsInConsumerGroups(consumerGroupDescription.get());


Suggested change

int partitionsInConsumerGroups = getPartitionsInConsumerGroups(consumerGroupDescription.get());

int consumerGroupPartitions = getConsumerGroupPartitions(consumerGroupDescription.get());

moscicky · 2023-09-08T10:10:10Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+            }
+        })).collect(toList());
+
+        waitForAllThreadsToEnd(futures);


how about creating a lists of tasks and the submiting them with ExecutorService.invokeAll()? This will automatically wait for all tasks to finish

moscicky · 2023-09-08T10:12:51Z

...main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringService.java

+
+    private boolean checkIfAllPartitionsAreAssigned(ConsumerGroupId consumerGroupId,
+                                                       KafkaTopics kafkaTopics) throws ExecutionException, InterruptedException {
+        Optional<ConsumerGroupDescription> consumerGroupDescription = getConsumerGroupDescription(consumerGroupId);


Have you considered batching requests to kafka? This should be faster than exeucting a request for each consume group

moscicky · 2023-09-08T10:17:40Z

...c/main/java/pl/allegro/tech/hermes/management/infrastructure/monitoring/MonitoringCache.java

+
+        List<Future<List<TopicAndSubscription>>> futures = splitSubscriptions.stream().map(it -> executorService.submit(() -> {
+            try {
+                List<MonitoringService> monitoringServices = monitoringServicesCreator.createMonitoringServices();


Maybe we can create monitoring services only once and cache them in this class?

# Conflicts: # build.gradle # hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/kafka/KafkaConfiguration.java

WojciechSova added the enhancement label Jan 18, 2023

WojciechSova temporarily deployed to ci January 18, 2023 10:14 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from d497146 to 0e9b256 Compare January 19, 2023 09:14

WojciechSova temporarily deployed to ci January 19, 2023 09:21 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from 0e9b256 to 0899c76 Compare January 19, 2023 12:22

WojciechSova temporarily deployed to ci January 19, 2023 12:23 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from ce51858 to f26a19d Compare January 19, 2023 13:31

WojciechSova had a problem deploying to ci January 19, 2023 13:32 — with GitHub Actions Error

WojciechSova force-pushed the consumergroup-monitoring-test branch from 5c55333 to 24c6952 Compare January 19, 2023 13:33

WojciechSova temporarily deployed to ci January 19, 2023 13:34 — with GitHub Actions Inactive

WojciechSova temporarily deployed to ci January 27, 2023 11:09 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from 9fca3be to 47bcf8c Compare January 27, 2023 12:27

WojciechSova temporarily deployed to ci January 27, 2023 12:27 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from 47bcf8c to 9db2e81 Compare January 27, 2023 12:51

WojciechSova temporarily deployed to ci January 27, 2023 12:51 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from 9db2e81 to b2172ae Compare January 27, 2023 13:36

WojciechSova had a problem deploying to ci January 27, 2023 13:37 — with GitHub Actions Failure

WojciechSova temporarily deployed to ci January 27, 2023 13:42 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from b2172ae to 8171e23 Compare January 30, 2023 09:45

WojciechSova temporarily deployed to ci January 30, 2023 10:31 — with GitHub Actions Inactive

WojciechSova force-pushed the consumergroup-monitoring-test branch from 8171e23 to 4e3cc26 Compare January 30, 2023 12:08

WojciechSova had a problem deploying to ci January 30, 2023 14:07 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 14:13 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 14:22 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 14:26 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 14:47 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 15:01 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 15:06 — with GitHub Actions Failure

WojciechSova had a problem deploying to ci January 30, 2023 15:19 — with GitHub Actions Failure

WojciechSova temporarily deployed to ci January 30, 2023 16:42 — with GitHub Actions Inactive

faderskd reviewed Feb 17, 2023

View reviewed changes

faderskd reviewed Mar 9, 2023

View reviewed changes

WojciechSova added 9 commits June 19, 2023 10:55

Test timings for consumergroup monitoring

ee01501

Test two threads

21c6592

Test consumer group monitoring

121295a

Refactor

55b756c

Add documentation

84b24f7

Refactor

b298422

Make corrections

8508a28

Set default properties

1d8a41e

PR fixes and added tests

1715077

faderskd force-pushed the consumergroup-monitoring-test branch from 341497b to 1715077 Compare June 19, 2023 08:55

WojciechSova and others added 3 commits June 19, 2023 16:10

Fix pr

9154378

Merge branch 'master' into consumergroup-monitoring-test

3c4a6b5

# Conflicts: # hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/ManagementConfiguration.java

Fix pr

f2bd48e

moscicky requested changes Sep 8, 2023

View reviewed changes

moscicky reviewed Sep 8, 2023

View reviewed changes

WojciechSova added 3 commits March 8, 2024 08:51

Merge branch 'master' into consumergroup-monitoring-test

31bc8e8

# Conflicts: # build.gradle # hermes-management/src/main/java/pl/allegro/tech/hermes/management/config/kafka/KafkaConfiguration.java

Fix PR Part 1 (only nitpicks)

69f0588

Checkstyle

0040e00

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Monitoring consumer groups #1650

Monitoring consumer groups #1650

WojciechSova commented Jan 18, 2023

faderskd Feb 17, 2023 •

edited

Loading

faderskd Feb 17, 2023

faderskd Feb 17, 2023

faderskd Mar 9, 2023

faderskd Mar 9, 2023 •

edited

Loading

WojciechSova May 9, 2023

moscicky Aug 8, 2023

faderskd Mar 9, 2023

faderskd Mar 9, 2023

faderskd Mar 9, 2023 •

edited

Loading

faderskd Mar 9, 2023

faderskd Mar 9, 2023

faderskd commented Mar 9, 2023

moscicky Jul 31, 2023

WojciechSova Sep 8, 2023

moscicky Jul 31, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Aug 8, 2023

moscicky Sep 8, 2023

moscicky Sep 8, 2023 •

edited

Loading

moscicky Sep 8, 2023

	int partitionsInConsumerGroups = getPartitionsInConsumerGroups(consumerGroupDescription.get());
	int consumerGroupPartitions = getConsumerGroupPartitions(consumerGroupDescription.get());

Monitoring consumer groups #1650

Are you sure you want to change the base?

Monitoring consumer groups #1650

Conversation

WojciechSova commented Jan 18, 2023

faderskd Feb 17, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

faderskd Mar 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

faderskd Mar 9, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

faderskd commented Mar 9, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

moscicky Sep 8, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

faderskd Feb 17, 2023 •

edited

Loading

faderskd Mar 9, 2023 •

edited

Loading

faderskd Mar 9, 2023 •

edited

Loading

moscicky Sep 8, 2023 •

edited

Loading