Skyeden-3020 | retransimission with detached consumers (#1938)

* SKYEDEN-3020 | Retransmission with detached consumers * SKYEDEN-3020 | Refactor * SKYEDEN-3020 | Refactor * SKYEDEN-3271 | comment + removed move offsets to the end from the subscription diagnostic tab + improved retransmission UI * SKYEDEN-3020 | changes related to PR + frontend tests * SKYEDEN-3020 | resolve conflicts * SKYEDEN-3020 | changes related to CR - added toString for PartitionOffset class
allegro · Jan 10, 2025 · f551c5b · f551c5b
1 parent da5a91f
commit f551c5b
Show file tree

Hide file tree

Showing 28 changed files with 407 additions and 325 deletions.
diff --git a/hermes-api/src/main/java/pl/allegro/tech/hermes/api/ConsumerGroup.java b/hermes-api/src/main/java/pl/allegro/tech/hermes/api/ConsumerGroup.java
@@ -41,6 +41,14 @@ public Set<ConsumerGroupMember> getMembers() {
     return members;
   }
 
+  public boolean isStable() {
+    return state.equals("Stable");
+  }
+
+  public boolean isEmpty() {
+    return state.equals("Empty");
+  }
+
   @Override
   public boolean equals(Object o) {
     if (this == o) {

diff --git a/hermes-common/src/main/java/pl/allegro/tech/hermes/common/kafka/offset/PartitionOffset.java b/hermes-common/src/main/java/pl/allegro/tech/hermes/common/kafka/offset/PartitionOffset.java
@@ -52,4 +52,16 @@ public boolean equals(Object obj) {
   public int hashCode() {
     return Objects.hash(topic, partition, offset);
   }
+
+  @Override
+  public String toString() {
+    return "PartitionOffset{"
+        + "topic="
+        + topic
+        + ", partition="
+        + partition
+        + ", offset="
+        + offset
+        + '}';
+  }
 }
diff --git a/hermes-console/json-server/server.ts b/hermes-console/json-server/server.ts
@@ -20,10 +20,6 @@ server.post('/query/subscriptions', (req, res) => {
   res.jsonp(subscriptions);
 });
 
-server.post('/topics/*/subscriptions/*/moveOffsetsToTheEnd', (req, res) => {
-  res.sendStatus(200);
-});
-
 server.post('/topicSubscriptions', (req, res) => {
   res.sendStatus(200);
 });
@@ -83,7 +79,9 @@ server.post('/offline-retransmission/tasks', (req, res) => {
 server.put(
   '/topics/:topic/subscriptions/:subscroption/retransmission',
   (req, res) => {
-    res.sendStatus(200);
+    setTimeout(() => {
+      res.sendStatus(200);
+    }, 2000);
   },
 );
 

diff --git a/hermes-console/src/api/hermes-client/index.ts b/hermes-console/src/api/hermes-client/index.ts
@@ -310,15 +310,6 @@ export function fetchDashboardUrl(path: string): ResponsePromise<DashboardUrl> {
   return axios.get<DashboardUrl>(path);
 }
 
-export function moveSubscriptionOffsets(
-  topicName: string,
-  subscription: string,
-): ResponsePromise<null> {
-  return axios.post<null>(
-    `/topics/${topicName}/subscriptions/${subscription}/moveOffsetsToTheEnd`,
-  );
-}
-
 export function removeTopic(topic: String): ResponsePromise<void> {
   return axios.delete(`/topics/${topic}`);
 }

diff --git a/hermes-console/src/composables/consumer-groups/use-consumer-groups/useConsumerGroups.spec.ts b/hermes-console/src/composables/consumer-groups/use-consumer-groups/useConsumerGroups.spec.ts
@@ -3,14 +3,9 @@ import { createTestingPinia } from '@pinia/testing';
 import { dummyConsumerGroups } from '@/dummy/consumerGroups';
 import { dummySubscription } from '@/dummy/subscription';
 import { dummyTopic } from '@/dummy/topic';
-import {
-  expectNotificationDispatched,
-  notificationStoreSpy,
-} from '@/utils/test-utils';
 import {
   fetchConsumerGroupsErrorHandler,
   fetchConsumerGroupsHandler,
-  moveSubscriptionOffsetsHandler,
 } from '@/mocks/handlers';
 import { setActivePinia } from 'pinia';
 import { setupServer } from 'msw/node';
@@ -81,56 +76,4 @@ describe('useConsumerGroups', () => {
       expect(error.value.fetchConsumerGroups).not.toBeNull();
     });
   });
-
-  it('should show message that moving offsets was successful', async () => {
-    // given
-    server.use(
-      moveSubscriptionOffsetsHandler({
-        topicName,
-        subscriptionName,
-        statusCode: 200,
-      }),
-    );
-    server.listen();
-    const notificationStore = notificationStoreSpy();
-
-    const { moveOffsets } = useConsumerGroups(topicName, subscriptionName);
-
-    // when
-    moveOffsets();
-
-    // then
-    await waitFor(() => {
-      expectNotificationDispatched(notificationStore, {
-        type: 'success',
-        title: 'notifications.subscriptionOffsets.move.success',
-      });
-    });
-  });
-
-  it('should show message that moving offsets was unsuccessful', async () => {
-    // given
-    server.use(
-      moveSubscriptionOffsetsHandler({
-        topicName,
-        subscriptionName,
-        statusCode: 500,
-      }),
-    );
-    server.listen();
-
-    const notificationStore = notificationStoreSpy();
-    const { moveOffsets } = useConsumerGroups(topicName, subscriptionName);
-
-    // when
-    moveOffsets();
-
-    // then
-    await waitFor(() => {
-      expectNotificationDispatched(notificationStore, {
-        type: 'error',
-        title: 'notifications.subscriptionOffsets.move.failure',
-      });
-    });
-  });
 });
diff --git a/hermes-console/src/composables/consumer-groups/use-consumer-groups/useConsumerGroups.ts b/hermes-console/src/composables/consumer-groups/use-consumer-groups/useConsumerGroups.ts
@@ -1,17 +1,10 @@
-import { dispatchErrorNotification } from '@/utils/notification-utils';
-import {
-  fetchConsumerGroups as getConsumerGroups,
-  moveSubscriptionOffsets,
-} from '@/api/hermes-client';
+import { fetchConsumerGroups as getConsumerGroups } from '@/api/hermes-client';
 import { ref } from 'vue';
-import { useGlobalI18n } from '@/i18n';
-import { useNotificationsStore } from '@/store/app-notifications/useAppNotifications';
 import type { ConsumerGroup } from '@/api/consumer-group';
 import type { Ref } from 'vue';
 
 export interface UseConsumerGroups {
   consumerGroups: Ref<ConsumerGroup[] | undefined>;
-  moveOffsets: () => void;
   loading: Ref<boolean>;
   error: Ref<UseConsumerGroupsErrors>;
 }
@@ -43,36 +36,10 @@ export function useConsumerGroups(
     }
   };
 
-  const moveOffsets = async () => {
-    const notificationsStore = useNotificationsStore();
-    try {
-      await moveSubscriptionOffsets(topicName, subscriptionName);
-      await notificationsStore.dispatchNotification({
-        title: useGlobalI18n().t(
-          'notifications.subscriptionOffsets.move.success',
-          {
-            subscriptionName,
-          },
-        ),
-        text: '',
-        type: 'success',
-      });
-    } catch (e: any) {
-      await dispatchErrorNotification(
-        e,
-        notificationsStore,
-        useGlobalI18n().t('notifications.subscriptionOffsets.move.failure', {
-          subscriptionName,
-        }),
-      );
-    }
-  };
-
   fetchConsumerGroups();
 
   return {
     consumerGroups,
-    moveOffsets,
     loading,
     error,
   };

diff --git a/hermes-console/src/composables/subscription/use-subscription/useSubscription.spec.ts b/hermes-console/src/composables/subscription/use-subscription/useSubscription.spec.ts
@@ -1,4 +1,4 @@
-import { afterEach } from 'vitest';
+import { afterEach, expect } from 'vitest';
 import {
   createRetransmissionHandler,
   fetchSubscriptionErrorHandler,
@@ -18,7 +18,6 @@ import {
   dummySubscriptionHealth,
   dummySubscriptionMetrics,
 } from '@/dummy/subscription';
-import { expect } from 'vitest';
 import {
   expectNotificationDispatched,
   notificationStoreSpy,
@@ -391,6 +390,39 @@ describe('useSubscription', () => {
     });
   });
 
+  [200, 500].forEach((statusCode) => {
+    it(`should correctly manage the state of retransmission with status code ${statusCode}`, async () => {
+      // given
+      server.use(
+        createRetransmissionHandler({
+          topicName: dummySubscription.topicName,
+          subscriptionName: dummySubscription.name,
+          statusCode,
+          delayMs: 100,
+        }),
+      );
+      server.listen();
+
+      const { retransmitMessages, retransmitting } = useSubscription(
+        dummySubscription.topicName,
+        dummySubscription.name,
+      );
+
+      expect(retransmitting.value).toBeFalsy();
+
+      // when
+      retransmitMessages(new Date().toISOString());
+
+      // then
+      await waitFor(() => {
+        expect(retransmitting.value).toBeTruthy();
+      });
+      await waitFor(() => {
+        expect(retransmitting.value).toBeFalsy();
+      });
+    });
+  });
+
   it('should show message that skipping all messages was successful', async () => {
     // given
     server.use(
@@ -448,6 +480,39 @@ describe('useSubscription', () => {
       });
     });
   });
+
+  [200, 500].forEach((statusCode) => {
+    it(`should correctly manage the state of skipping all messages with status code ${statusCode}`, async () => {
+      // given
+      server.use(
+        createRetransmissionHandler({
+          topicName: dummySubscription.topicName,
+          subscriptionName: dummySubscription.name,
+          statusCode,
+          delayMs: 100,
+        }),
+      );
+      server.listen();
+
+      const { skipAllMessages, skippingAllMessages } = useSubscription(
+        dummySubscription.topicName,
+        dummySubscription.name,
+      );
+
+      expect(skippingAllMessages.value).toBeFalsy();
+
+      // when
+      skipAllMessages();
+
+      // then
+      await waitFor(() => {
+        expect(skippingAllMessages.value).toBeTruthy();
+      });
+      await waitFor(() => {
+        expect(skippingAllMessages.value).toBeFalsy();
+      });
+    });
+  });
 });
 
 function expectErrors(

diff --git a/hermes-console/src/composables/subscription/use-subscription/useSubscription.ts b/hermes-console/src/composables/subscription/use-subscription/useSubscription.ts
@@ -32,6 +32,8 @@ export interface UseSubscription {
   subscriptionLastUndeliveredMessage: Ref<SentMessageTrace | null>;
   trackingUrls: Ref<TrackingUrl[] | undefined>;
   loading: Ref<boolean>;
+  retransmitting: Ref<boolean>;
+  skippingAllMessages: Ref<boolean>;
   error: Ref<UseSubscriptionsErrors>;
   removeSubscription: () => Promise<boolean>;
   suspendSubscription: () => Promise<boolean>;
@@ -73,7 +75,8 @@ export function useSubscription(
     fetchSubscriptionLastUndeliveredMessage: null,
     getSubscriptionTrackingUrls: null,
   });
-
+  const retransmitting = ref(false);
+  const skippingAllMessages = ref(false);
   const fetchSubscription = async () => {
     try {
       loading.value = true;
@@ -233,6 +236,7 @@ export function useSubscription(
   };
 
   const retransmitMessages = async (from: string): Promise<boolean> => {
+    retransmitting.value = true;
     try {
       await retransmitSubscriptionMessages(topicName, subscriptionName, {
         retransmissionDate: from,
@@ -257,10 +261,13 @@ export function useSubscription(
         }),
       );
       return false;
+    } finally {
+      retransmitting.value = false;
     }
   };
 
   const skipAllMessages = async (): Promise<boolean> => {
+    skippingAllMessages.value = true;
     const tomorrowDate = new Date();
     tomorrowDate.setDate(tomorrowDate.getDate() + 1);
     try {
@@ -290,6 +297,8 @@ export function useSubscription(
         ),
       );
       return false;
+    } finally {
+      skippingAllMessages.value = false;
     }
   };
 
@@ -305,6 +314,8 @@ export function useSubscription(
     subscriptionLastUndeliveredMessage,
     trackingUrls,
     loading,
+    retransmitting,
+    skippingAllMessages,
     error,
     removeSubscription,
     suspendSubscription,

diff --git a/hermes-console/src/i18n/en-US/index.ts b/hermes-console/src/i18n/en-US/index.ts
@@ -625,10 +625,6 @@ const en_US = {
       reason: 'Reason',
       timestamp: 'Timestamp',
     },
-    moveOffsets: {
-      tooltip: 'Move subscription offsets to the end',
-      button: 'MOVE OFFSETS',
-    },
   },
   search: {
     collection: {