Skyeden-3020 | retransimission with detached consumers

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) {

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);
   },
 );
 

hermes-console/src/api/hermes-client/index.ts

@@ -294,15 +294,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}`);
 }

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',
-      });
-    });
-  });
 });

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,
   };

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: {

hermes-console/src/mocks/handlers.ts

@@ -817,24 +817,6 @@ export const switchReadinessErrorHandler = ({
     });
   });
 
-export const moveSubscriptionOffsetsHandler = ({
-  topicName,
-  subscriptionName,
-  statusCode,
-}: {
-  topicName: string;
-  subscriptionName: string;
-  statusCode: number;
-}) =>
-  http.post(
-    `${url}/topics/${topicName}/subscriptions/${subscriptionName}/moveOffsetsToTheEnd`,
-    () => {
-      return new HttpResponse(undefined, {
-        status: statusCode,
-      });
-    },
-  );
-
 export const upsertTopicConstraintHandler = ({
   statusCode,
 }: {

hermes-console/src/views/admin/consumer-groups/ConsumerGroupsView.vue

@@ -12,7 +12,7 @@
   const params = route.params as Record<string, string>;
   const { subscriptionId, topicId, groupId } = params;
 
-  const { consumerGroups, moveOffsets, loading, error } = useConsumerGroups(
+  const { consumerGroups, loading, error } = useConsumerGroups(
     topicId,
     subscriptionId,
   );
@@ -64,14 +64,6 @@
           {{ $t('consumerGroups.title') }}
         </p>
       </v-col>
-      <v-col md="2" class="text-right">
-        <v-btn color="red" @click="moveOffsets">
-          {{ $t('subscription.moveOffsets.button') }}
-          <v-tooltip activator="parent" location="left">{{
-            $t('subscription.moveOffsets.tooltip')
-          }}</v-tooltip>
-        </v-btn>
-      </v-col>
     </v-row>
     <v-row dense v-if="consumerGroups">
       <v-col>

hermes-console/src/views/subscription/SubscriptionView.vue

@@ -102,8 +102,12 @@
     }
   }
 
-  const onRetransmit = async (fromDate: string) => {
-    await retransmitMessages(fromDate);
+  const onRetransmit = async (fromDate: string, onComplete: () => void) => {
+    await retransmitMessages(fromDate).finally(onComplete);
+  };
+
+  const onSkipAllMessages = async (onComplete: () => void) => {
+    await skipAllMessages().finally(onComplete);
   };
 
   const breadcrumbsItems = [
@@ -229,7 +233,7 @@
             :topic="topicId"
             :subscription="subscriptionId"
             @retransmit="onRetransmit"
-            @skipAllMessages="skipAllMessages"
+            @skipAllMessages="onSkipAllMessages"
           />
           <last-undelivered-message
             v-if="

hermes-console/src/views/subscription/manage-messages-card/ManageMessagesCard.vue

@@ -4,6 +4,7 @@
   import { useDialog } from '@/composables/dialog/use-dialog/useDialog';
   import { useI18n } from 'vue-i18n';
   import ConfirmationDialog from '@/components/confirmation-dialog/ConfirmationDialog.vue';
+  import LoadingSpinner from '@/components/loading-spinner/LoadingSpinner.vue';
 
   const { t } = useI18n();
 
@@ -13,8 +14,8 @@
   }>();
 
   const emit = defineEmits<{
-    retransmit: [fromDate: string];
-    skipAllMessages: [];
+    retransmit: [fromDate: string, onComplete: () => void];
+    skipAllMessages: [onComplete: () => void];
   }>();
 
   const retransmitFrom = ref(new Date().toISOString().slice(0, -5));
@@ -42,17 +43,20 @@
     disableActionButton: disableSkipAllMessagesActionButton,
   } = useDialog();
 
+  const retransmitting = ref(false);
+  const skippingAllMessages = ref(false);
+
   async function retransmit() {
     disableRetransmitActionButton();
-    emit('retransmit', `${retransmitFrom.value}Z`);
-    enableRetransmitActionButton();
+    retransmitting.value = true;
+    emit('retransmit', `${retransmitFrom.value}Z`, retransmissionCompleted);
     closeRetransmitDialog();
   }
 
   async function skipMessages() {
     disableSkipAllMessagesActionButton();
-    emit('skipAllMessages');
-    enableSkipAllMessagesActionButton();
+    skippingAllMessages.value = true;
+    emit('skipAllMessages', skippingAllMessagesCompleted);
     closeSkipAllMessagesDialog();
   }
 
@@ -68,6 +72,16 @@
       subscriptionFqn: subscriptionQualifiedName,
     }),
   );
+
+  function retransmissionCompleted() {
+    retransmitting.value = false;
+    enableRetransmitActionButton();
+  }
+
+  function skippingAllMessagesCompleted() {
+    skippingAllMessages.value = false;
+    enableSkipAllMessagesActionButton();
+  }
 </script>
 
 <template>
@@ -113,11 +127,15 @@
           </v-col>
           <v-col md="3">
             <v-btn
+              :disabled="retransmitting || skippingAllMessages"
               color="red"
               @click="openRetransmitDialog"
               data-testid="retransmitButton"
             >
-              {{ $t('subscription.manageMessagesCard.retransmitButton') }}
+              <loading-spinner v-if="retransmitting"></loading-spinner>
+              <span v-else>
+                {{ $t('subscription.manageMessagesCard.retransmitButton') }}
+              </span>
             </v-btn>
           </v-col>
         </v-row>
@@ -128,12 +146,16 @@
           {{ $t('subscription.manageMessagesCard.skipAllMessagesTitle') }}
         </p>
         <v-btn
+          :disabled="retransmitting || skippingAllMessages"
           color="red"
           class="mt-2"
           @click="openSkipAllMessagesDialog"
           data-testid="skipAllMessagesButton"
         >
-          {{ $t('subscription.manageMessagesCard.skipAllMessagesButton') }}
+          <loading-spinner v-if="skippingAllMessages"></loading-spinner>
+          <span v-else>
+            {{ $t('subscription.manageMessagesCard.skipAllMessagesButton') }}
+          </span>
         </v-btn>
       </v-card-item>
     </v-form>

hermes-management/src/main/java/pl/allegro/tech/hermes/management/api/SubscriptionsEndpoint.java

@@ -35,7 +35,6 @@
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.SubscriptionHealth;
 import pl.allegro.tech.hermes.api.SubscriptionMetrics;
-import pl.allegro.tech.hermes.api.SubscriptionName;
 import pl.allegro.tech.hermes.api.Topic;
 import pl.allegro.tech.hermes.api.TopicName;
 import pl.allegro.tech.hermes.management.api.auth.HermesSecurityAwareRequestUser;
@@ -273,7 +272,7 @@ public Response retransmit(
       @Context ContainerRequestContext requestContext) {
 
     MultiDCOffsetChangeSummary summary =
-        multiDCAwareService.retransmit(
+        subscriptionService.retransmit(
             topicService.getTopicDetails(TopicName.fromQualifiedName(qualifiedTopicName)),
             subscriptionName,
             offsetRetransmissionDate.getRetransmissionDate().toInstant().toEpochMilli(),
@@ -283,20 +282,6 @@ public Response retransmit(
     return Response.status(OK).entity(summary).build();
   }
 
-  @POST
-  @Consumes(APPLICATION_JSON)
-  @Produces(APPLICATION_JSON)
-  @RolesAllowed({Roles.ADMIN})
-  @Path("/{subscriptionName}/moveOffsetsToTheEnd")
-  public Response moveOffsetsToTheEnd(
-      @PathParam("topicName") String qualifiedTopicName,
-      @PathParam("subscriptionName") String subscriptionName) {
-    TopicName topicName = fromQualifiedName(qualifiedTopicName);
-    multiDCAwareService.moveOffsetsToTheEnd(
-        topicService.getTopicDetails(topicName), new SubscriptionName(subscriptionName, topicName));
-    return responseStatus(OK);
-  }
-
   @GET
   @Produces(APPLICATION_JSON)
   @Path("/{subscriptionName}/events/{messageId}/trace")