Merge branch 'master' into ISSUE_754_acl

.github/workflows/branch-deploy.yml

@@ -9,9 +9,9 @@ jobs:
     if: ${{ github.event.label.name == 'status/feature_testing' || github.event.label.name == 'status/feature_testing_public' }}
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v3
+      - uses: ./.github/workflows/build-template.yaml
         with:
-          ref: ${{ github.event.pull_request.head.sha }}
+          APP_VERSION: $GITHUB_SHA
       - name: get branch name
         id: extract_branch
         run: |
@@ -19,19 +19,6 @@ jobs:
           echo "tag=${tag}" >> $GITHUB_OUTPUT
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
-      - name: Set up JDK
-        uses: actions/setup-java@v3
-        with:
-          java-version: '17'
-          distribution: 'zulu'
-          cache: 'maven'
-      - name: Build
-        id: build
-        run: |
-          ./mvnw -B -ntp versions:set -DnewVersion=$GITHUB_SHA
-          ./mvnw -B -V -ntp clean package -Pprod -DskipTests
-          export VERSION=$(./mvnw -q -Dexec.executable=echo -Dexec.args='${project.version}' --non-recursive exec:exec)
-          echo "version=${VERSION}" >> $GITHUB_OUTPUT
       - name: Set up QEMU
         uses: docker/setup-qemu-action@v2
       - name: Set up Docker Buildx

.github/workflows/build-template.yml

@@ -0,0 +1,32 @@
+name: Maven build template
+on:
+  workflow_call:
+   inputs:
+    APP_VERSION:
+     required: true
+     type: string
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    outputs:
+      version: ${{steps.build.outputs.version}}
+    steps:
+      - uses: actions/checkout@v3
+        with:
+          ref: ${{ github.event.pull_request.head.sha }}
+      - run: |
+          git config user.name github-actions
+          git config user.email github-actions@github.com
+      - name: Set up JDK
+        uses: actions/setup-java@v3
+        with:
+          java-version: '17'
+          distribution: 'zulu'
+          cache: 'maven'
+      - name: Build
+        id: build
+        run: |
+          ./mvnw -B -ntp versions:set -DnewVersion=${{ inputs.APP_VERSION }}
+          ./mvnw -B -V -ntp clean package -Pprod -DskipTests
+          export VERSION=$(./mvnw -q -Dexec.executable=echo -Dexec.args='${project.version}' --non-recursive exec:exec)
+          echo "version=${VERSION}" >> $GITHUB_OUTPUT

.github/workflows/terraform-deploy.yml

@@ -26,7 +26,7 @@ jobs:
           echo "Terraform will be triggered in this dir $TF_DIR"
 
       - name: Configure AWS credentials for Kafka-UI account
-        uses: aws-actions/configure-aws-credentials@v1
+        uses: aws-actions/configure-aws-credentials@v1-node16
         with:
           aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
           aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java

@@ -25,6 +25,8 @@ public class ClustersProperties {
 
   List<Cluster> clusters = new ArrayList<>();
 
+  String internalTopicPrefix;
+
   @Data
   public static class Cluster {
     String name;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java

@@ -1,9 +1,11 @@
 package com.provectus.kafka.ui.model;
 
+import com.provectus.kafka.ui.config.ClustersProperties;
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import lombok.Builder;
 import lombok.Data;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -14,6 +16,8 @@ import org.apache.kafka.common.TopicPartition;
 @Builder(toBuilder = true)
 public class InternalTopic {
 
+  ClustersProperties clustersProperties;
+
   // from TopicDescription
   private final String name;
   private final boolean internal;
@@ -40,9 +44,17 @@ public class InternalTopic {
                                    List<ConfigEntry> configs,
                                    InternalPartitionsOffsets partitionsOffsets,
                                    Metrics metrics,
-                                   InternalLogDirStats logDirInfo) {
+                                   InternalLogDirStats logDirInfo,
+                                   @Nullable String internalTopicPrefix) {
     var topic = InternalTopic.builder();
-    topic.internal(topicDescription.isInternal());
+
+    internalTopicPrefix = internalTopicPrefix == null || internalTopicPrefix.isEmpty()
+        ? "_"
+        : internalTopicPrefix;
+
+    topic.internal(
+        topicDescription.isInternal() || topicDescription.name().startsWith(internalTopicPrefix)
+    );
     topic.name(topicDescription.name());
 
     List<InternalPartition> partitions = topicDescription.partitions().stream()
@@ -56,10 +68,10 @@ public class InternalTopic {
           List<InternalReplica> replicas = partition.replicas().stream()
               .map(r ->
                   InternalReplica.builder()
-                    .broker(r.id())
-                    .inSync(partition.isr().contains(r))
-                    .leader(partition.leader() != null && partition.leader().id() == r.id())
-                    .build())
+                      .broker(r.id())
+                      .inSync(partition.isr().contains(r))
+                      .leader(partition.leader() != null && partition.leader().id() == r.id())
+                      .build())
               .collect(Collectors.toList());
           partitionDto.replicas(replicas);
 
@@ -79,7 +91,7 @@ public class InternalTopic {
 
           return partitionDto.build();
         })
-        .collect(Collectors.toList());
+        .toList();
 
     topic.partitions(partitions.stream().collect(
         Collectors.toMap(InternalPartition::getPartition, t -> t)));

kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java

@@ -3,6 +3,7 @@ package com.provectus.kafka.ui.service;
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
 
+import com.provectus.kafka.ui.config.ClustersProperties;
 import com.provectus.kafka.ui.exception.TopicMetadataException;
 import com.provectus.kafka.ui.exception.TopicNotFoundException;
 import com.provectus.kafka.ui.exception.TopicRecreationException;
@@ -52,6 +53,7 @@ public class TopicsService {
 
   private final AdminClientService adminClientService;
   private final StatisticsCache statisticsCache;
+  private final ClustersProperties clustersProperties;
   @Value("${topic.recreate.maxRetries:15}")
   private int recreateMaxRetries;
   @Value("${topic.recreate.delay.seconds:1}")
@@ -127,7 +129,8 @@ public class TopicsService {
             configs.getOrDefault(t, List.of()),
             partitionsOffsets,
             metrics,
-            logDirInfo
+            logDirInfo,
+            clustersProperties.getInternalTopicPrefix()
         ))
         .collect(toList());
   }
@@ -459,7 +462,9 @@ public class TopicsService {
                     stats.getTopicConfigs().getOrDefault(topicName, List.of()),
                     InternalPartitionsOffsets.empty(),
                     stats.getMetrics(),
-                    stats.getLogDirInfo()))
+                    stats.getLogDirInfo(),
+                    clustersProperties.getInternalTopicPrefix()
+                    ))
             .collect(toList())
         );
   }

kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java

@@ -69,7 +69,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -95,7 +95,7 @@ class TopicsServicePaginationTest {
         .map(Objects::toString)
         .map(name -> new TopicDescription(name, false, List.of()))
         .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-            Metrics.empty(), InternalLogDirStats.empty()))
+            Metrics.empty(), InternalLogDirStats.empty(), "_"))
         .collect(Collectors.toMap(InternalTopic::getName, Function.identity()));
     init(internalTopics);
 
@@ -122,7 +122,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -141,7 +141,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -160,7 +160,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, Integer.parseInt(name) % 10 == 0, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -181,7 +181,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, Integer.parseInt(name) % 5 == 0, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -202,7 +202,7 @@ class TopicsServicePaginationTest {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -224,7 +224,7 @@ class TopicsServicePaginationTest {
                     new TopicPartitionInfo(p, null, List.of(), List.of()))
                 .collect(Collectors.toList())))
         .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), InternalPartitionsOffsets.empty(),
-            Metrics.empty(), InternalLogDirStats.empty()))
+            Metrics.empty(), InternalLogDirStats.empty(), "_"))
         .collect(Collectors.toMap(InternalTopic::getName, Function.identity()));
 
     init(internalTopics);

kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java

@@ -6,7 +6,6 @@ import com.codeborne.selenide.SelenideElement;
 import com.provectus.kafka.ui.pages.BasePage;
 import io.qameta.allure.Step;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -175,6 +174,12 @@ public class TopicsList extends BasePage {
                 .findFirst().orElseThrow();
     }
 
+    @Step
+    public TopicGridItem getAnyNonInternalTopic() {
+        return getNonInternalTopics().stream()
+                .findAny().orElseThrow();
+    }
+
     @Step
     public List<TopicGridItem> getNonInternalTopics() {
         return initGridItems().stream()
@@ -207,8 +212,7 @@ public class TopicsList extends BasePage {
         public boolean isInternal() {
             boolean internal = false;
             try {
-                element.$x("./td[2]/a/span").shouldBe(visible, Duration.ofMillis(500));
-                internal = true;
+                internal = element.$x("./td[2]/a/span").isDisplayed();
             } catch (Throwable ignored) {
             }
             return internal;

kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java

@@ -52,7 +52,8 @@ public class TopicsTest extends BaseTest {
             .setMaxSizeOnDisk(NOT_SET);
     private static final Topic TOPIC_FOR_CHECK_FILTERS = new Topic()
             .setName("topic-for-check-filters-" + randomAlphabetic(5));
-    private static final Topic TOPIC_FOR_DELETE = new Topic().setName("topic-to-delete-" + randomAlphabetic(5));
+    private static final Topic TOPIC_FOR_DELETE = new Topic()
+            .setName("topic-to-delete-" + randomAlphabetic(5));
     private static final List<Topic> TOPIC_LIST = new ArrayList<>();
 
     @BeforeClass(alwaysRun = true)
@@ -89,11 +90,11 @@ public class TopicsTest extends BaseTest {
     void checkAvailableOperations() {
         navigateToTopics();
         topicsList
-                .getTopicItem("my_ksql_1ksql_processing_log")
+                .getTopicItem(TOPIC_TO_UPDATE_AND_DELETE.getName())
                 .selectItem(true);
         verifyElementsCondition(topicsList.getActionButtons(), Condition.enabled);
         topicsList
-                .getTopicItem("_confluent-ksql-my_ksql_1_command_topic")
+                .getTopicItem(TOPIC_FOR_CHECK_FILTERS.getName())
                 .selectItem(true);
         Assert.assertFalse(topicsList.isCopySelectedTopicBtnEnabled(), "isCopySelectedTopicBtnEnabled()");
     }
@@ -456,7 +457,7 @@ public class TopicsTest extends BaseTest {
                 .setNumberOfPartitions(1);
         navigateToTopics();
         topicsList
-                .getTopicItem("_schemas")
+                .getAnyNonInternalTopic()
                 .selectItem(true)
                 .clickCopySelectedTopicBtn();
         topicCreateEditForm

kafka-ui-react-app/src/components/Topics/List/ActionsCell.tsx

@@ -1,20 +1,17 @@
 import React from 'react';
 import { Action, CleanUpPolicy, Topic, ResourceType } from 'generated-sources';
 import { CellContext } from '@tanstack/react-table';
-import { useAppDispatch } from 'lib/hooks/redux';
 import ClusterContext from 'components/contexts/ClusterContext';
 import { ClusterNameRoute } from 'lib/paths';
 import useAppParams from 'lib/hooks/useAppParams';
-import { clearTopicMessages } from 'redux/reducers/topicMessages/topicMessagesSlice';
 import {
   Dropdown,
   DropdownItem,
   DropdownItemHint,
 } from 'components/common/Dropdown';
-import { useQueryClient } from '@tanstack/react-query';
 import {
-  topicKeys,
   useDeleteTopic,
+  useClearTopicMessages,
   useRecreateTopic,
 } from 'lib/hooks/api/topics';
 import { ActionDropdownItem } from 'components/common/ActionComponent';
@@ -24,20 +21,16 @@ const ActionsCell: React.FC<CellContext<Topic, unknown>> = ({ row }) => {
 
   const { isReadOnly, isTopicDeletionAllowed } =
     React.useContext(ClusterContext);
-  const dispatch = useAppDispatch();
   const { clusterName } = useAppParams<ClusterNameRoute>();
-  const queryClient = useQueryClient();
 
+  const clearMessages = useClearTopicMessages(clusterName);
   const deleteTopic = useDeleteTopic(clusterName);
   const recreateTopic = useRecreateTopic({ clusterName, topicName: name });
 
   const disabled = internal || isReadOnly;
 
   const clearTopicMessagesHandler = async () => {
-    await dispatch(
-      clearTopicMessages({ clusterName, topicName: name })
-    ).unwrap();
-    return queryClient.invalidateQueries(topicKeys.all(clusterName));
+    await clearMessages.mutateAsync(name);
   };
 
   const isCleanupDisabled = cleanUpPolicy !== CleanUpPolicy.DELETE;

kafka-ui-react-app/src/components/Topics/List/BatchActionsBar.tsx

@@ -3,11 +3,13 @@ import { Row } from '@tanstack/react-table';
 import { Action, Topic, ResourceType } from 'generated-sources';
 import useAppParams from 'lib/hooks/useAppParams';
 import { ClusterName } from 'redux/interfaces';
-import { topicKeys, useDeleteTopic } from 'lib/hooks/api/topics';
+import {
+  topicKeys,
+  useClearTopicMessages,
+  useDeleteTopic,
+} from 'lib/hooks/api/topics';
 import { useConfirm } from 'lib/hooks/useConfirm';
 import { Button } from 'components/common/Button/Button';
-import { useAppDispatch } from 'lib/hooks/redux';
-import { clearTopicMessages } from 'redux/reducers/topicMessages/topicMessagesSlice';
 import { clusterTopicCopyRelativePath } from 'lib/paths';
 import { useQueryClient } from '@tanstack/react-query';
 import { ActionCanButton } from 'components/common/ActionComponent';
@@ -25,11 +27,14 @@ const BatchActionsbar: React.FC<BatchActionsbarProps> = ({
 }) => {
   const { clusterName } = useAppParams<{ clusterName: ClusterName }>();
   const confirm = useConfirm();
-  const dispatch = useAppDispatch();
   const deleteTopic = useDeleteTopic(clusterName);
   const selectedTopics = rows.map(({ original }) => original.name);
   const client = useQueryClient();
 
+  const clearMessages = useClearTopicMessages(clusterName);
+  const clearTopicMessagesHandler = async (topicName: Topic['name']) => {
+    await clearMessages.mutateAsync(topicName);
+  };
   const deleteTopicsHandler = () => {
     confirm('Are you sure you want to remove selected topics?', async () => {
       try {
@@ -50,7 +55,7 @@ const BatchActionsbar: React.FC<BatchActionsbarProps> = ({
         try {
           await Promise.all(
             selectedTopics.map((topicName) =>
-              dispatch(clearTopicMessages({ clusterName, topicName })).unwrap()
+              clearTopicMessagesHandler(topicName)
             )
           );
           resetRowSelection();

kafka-ui-react-app/src/components/Topics/List/__tests__/TopicTable.spec.tsx

@@ -6,16 +6,15 @@ import { externalTopicPayload, topicsPayload } from 'lib/fixtures/topics';
 import ClusterContext from 'components/contexts/ClusterContext';
 import userEvent from '@testing-library/user-event';
 import {
+  useClearTopicMessages,
   useDeleteTopic,
   useRecreateTopic,
   useTopics,
 } from 'lib/hooks/api/topics';
 import TopicTable from 'components/Topics/List/TopicTable';
 import { clusterTopicsPath } from 'lib/paths';
-import { useAppDispatch } from 'lib/hooks/redux';
 
 const clusterName = 'test-cluster';
-const unwrapMock = jest.fn();
 
 jest.mock('lib/hooks/redux', () => ({
   ...jest.requireActual('lib/hooks/redux'),
@@ -29,22 +28,24 @@ jest.mock('lib/hooks/api/topics', () => ({
   useDeleteTopic: jest.fn(),
   useRecreateTopic: jest.fn(),
   useTopics: jest.fn(),
+  useClearTopicMessages: jest.fn(),
 }));
 
 const deleteTopicMock = jest.fn();
 const recreateTopicMock = jest.fn();
+const clearTopicMessages = jest.fn();
 
 describe('TopicTable Components', () => {
   beforeEach(() => {
     (useDeleteTopic as jest.Mock).mockImplementation(() => ({
       mutateAsync: deleteTopicMock,
     }));
+    (useClearTopicMessages as jest.Mock).mockImplementation(() => ({
+      mutateAsync: clearTopicMessages,
+    }));
     (useRecreateTopic as jest.Mock).mockImplementation(() => ({
       mutateAsync: recreateTopicMock,
     }));
-    (useAppDispatch as jest.Mock).mockImplementation(() => () => ({
-      unwrap: unwrapMock,
-    }));
   });
 
   const renderComponent = (
@@ -185,9 +186,9 @@ describe('TopicTable Components', () => {
             ).toBeInTheDocument();
             const confirmBtn = getButtonByName('Confirm');
             expect(confirmBtn).toBeInTheDocument();
-            expect(unwrapMock).not.toHaveBeenCalled();
+            expect(clearTopicMessages).not.toHaveBeenCalled();
             await userEvent.click(confirmBtn);
-            expect(unwrapMock).toHaveBeenCalledTimes(2);
+            expect(clearTopicMessages).toHaveBeenCalledTimes(2);
             expect(screen.getAllByRole('checkbox')[1]).not.toBeChecked();
             expect(screen.getAllByRole('checkbox')[2]).not.toBeChecked();
           });
@@ -282,7 +283,7 @@ describe('TopicTable Components', () => {
           await userEvent.click(
             screen.getByRole('button', { name: 'Confirm' })
           );
-          expect(unwrapMock).toHaveBeenCalled();
+          expect(clearTopicMessages).toHaveBeenCalled();
         });
       });
 

kafka-ui-react-app/src/components/Topics/Topic/Messages/Filters/Filters.tsx

@@ -125,10 +125,10 @@ const Filters: React.FC<FiltersProps> = ({
     getTimestampFromSeekToParam(searchParams)
   );
   const [keySerde, setKeySerde] = React.useState<string>(
-    searchParams.get('keySerde') as string
+    searchParams.get('keySerde') || ''
   );
   const [valueSerde, setValueSerde] = React.useState<string>(
-    searchParams.get('valueSerde') as string
+    searchParams.get('valueSerde') || ''
   );
 
   const [savedFilters, setSavedFilters] = React.useState<MessageFilters[]>(
@@ -206,8 +206,8 @@ const Filters: React.FC<FiltersProps> = ({
       limit: PER_PAGE,
       page: page || 0,
       seekDirection,
-      keySerde: keySerde || (searchParams.get('keySerde') as string),
-      valueSerde: valueSerde || (searchParams.get('valueSerde') as string),
+      keySerde: keySerde || searchParams.get('keySerde') || '',
+      valueSerde: valueSerde || searchParams.get('valueSerde') || '',
     };
 
     if (isSeekTypeControlVisible) {

kafka-ui-react-app/src/components/Topics/Topic/Overview/ActionsCell.tsx

@@ -1,13 +1,11 @@
 import React from 'react';
 import { Action, Partition, ResourceType } from 'generated-sources';
 import { CellContext } from '@tanstack/react-table';
-import { useAppDispatch } from 'lib/hooks/redux';
 import ClusterContext from 'components/contexts/ClusterContext';
 import { RouteParamsClusterTopic } from 'lib/paths';
 import useAppParams from 'lib/hooks/useAppParams';
-import { clearTopicMessages } from 'redux/reducers/topicMessages/topicMessagesSlice';
 import { Dropdown } from 'components/common/Dropdown';
-import { useTopicDetails } from 'lib/hooks/api/topics';
+import { useClearTopicMessages, useTopicDetails } from 'lib/hooks/api/topics';
 import { ActionDropdownItem } from 'components/common/ActionComponent';
 
 const ActionsCell: React.FC<CellContext<Partition, unknown>> = ({ row }) => {
@@ -15,12 +13,11 @@ const ActionsCell: React.FC<CellContext<Partition, unknown>> = ({ row }) => {
   const { data } = useTopicDetails({ clusterName, topicName });
   const { isReadOnly } = React.useContext(ClusterContext);
   const { partition } = row.original;
-  const dispatch = useAppDispatch();
+
+  const clearMessages = useClearTopicMessages(clusterName, [partition]);
 
   const clearTopicMessagesHandler = async () => {
-    await dispatch(
-      clearTopicMessages({ clusterName, topicName, partitions: [partition] })
-    ).unwrap();
+    await clearMessages.mutateAsync(topicName);
   };
   const disabled =
     data?.internal || isReadOnly || data?.cleanUpPolicy !== 'DELETE';

kafka-ui-react-app/src/components/Topics/Topic/Overview/__test__/Overview.spec.tsx

@@ -8,8 +8,7 @@ import ClusterContext from 'components/contexts/ClusterContext';
 import userEvent from '@testing-library/user-event';
 import { clusterTopicPath } from 'lib/paths';
 import { Replica } from 'components/Topics/Topic/Overview/Overview.styled';
-import { useTopicDetails } from 'lib/hooks/api/topics';
-import { useAppDispatch } from 'lib/hooks/redux';
+import { useClearTopicMessages, useTopicDetails } from 'lib/hooks/api/topics';
 import {
   externalTopicPayload,
   internalTopicPayload,
@@ -26,14 +25,10 @@ const defaultContextValues = {
 
 jest.mock('lib/hooks/api/topics', () => ({
   useTopicDetails: jest.fn(),
+  useClearTopicMessages: jest.fn(),
 }));
 
-const unwrapMock = jest.fn();
-
-jest.mock('lib/hooks/redux', () => ({
-  ...jest.requireActual('lib/hooks/redux'),
-  useAppDispatch: jest.fn(),
-}));
+const clearTopicMessage = jest.fn();
 
 describe('Overview', () => {
   const renderComponent = (
@@ -43,6 +38,9 @@ describe('Overview', () => {
     (useTopicDetails as jest.Mock).mockImplementation(() => ({
       data: topic,
     }));
+    (useClearTopicMessages as jest.Mock).mockImplementation(() => ({
+      mutateAsync: clearTopicMessage,
+    }));
     const path = clusterTopicPath(clusterName, topicName);
     return render(
       <WithRoute path={clusterTopicPath()}>
@@ -54,12 +52,6 @@ describe('Overview', () => {
     );
   };
 
-  beforeEach(() => {
-    (useAppDispatch as jest.Mock).mockImplementation(() => () => ({
-      unwrap: unwrapMock,
-    }));
-  });
-
   it('at least one replica was rendered', () => {
     renderComponent();
     expect(screen.getByLabelText('replica-info')).toBeInTheDocument();
@@ -136,7 +128,7 @@ describe('Overview', () => {
 
       const clearMessagesButton = screen.getByText('Clear Messages');
       await userEvent.click(clearMessagesButton);
-      expect(unwrapMock).toHaveBeenCalledTimes(1);
+      expect(clearTopicMessage).toHaveBeenCalledTimes(1);
     });
   });
 

kafka-ui-react-app/src/components/Topics/Topic/Topic.tsx

@@ -21,14 +21,12 @@ import { useAppDispatch } from 'lib/hooks/redux';
 import useAppParams from 'lib/hooks/useAppParams';
 import { Dropdown, DropdownItemHint } from 'components/common/Dropdown';
 import {
+  useClearTopicMessages,
   useDeleteTopic,
   useRecreateTopic,
   useTopicDetails,
 } from 'lib/hooks/api/topics';
-import {
-  clearTopicMessages,
-  resetTopicMessages,
-} from 'redux/reducers/topicMessages/topicMessagesSlice';
+import { resetTopicMessages } from 'redux/reducers/topicMessages/topicMessagesSlice';
 import { Action, CleanUpPolicy, ResourceType } from 'generated-sources';
 import PageLoader from 'components/common/PageLoader/PageLoader';
 import SlidingSidebar from 'components/common/SlidingSidebar';
@@ -69,9 +67,11 @@ const Topic: React.FC = () => {
       dispatch(resetTopicMessages());
     };
   }, []);
-
+  const clearMessages = useClearTopicMessages(clusterName);
+  const clearTopicMessagesHandler = async () => {
+    await clearMessages.mutateAsync(topicName);
+  };
   const canCleanup = data?.cleanUpPolicy === CleanUpPolicy.DELETE;
-
   return (
     <>
       <PageHeading
@@ -110,9 +110,7 @@ const Topic: React.FC = () => {
           </ActionDropdownItem>
 
           <ActionDropdownItem
-            onClick={() =>
-              dispatch(clearTopicMessages({ clusterName, topicName })).unwrap()
-            }
+            onClick={clearTopicMessagesHandler}
             confirm="Are you sure want to clear topic messages?"
             disabled={!canCleanup}
             danger

kafka-ui-react-app/src/components/Topics/Topic/__test__/Topic.spec.tsx

@@ -16,6 +16,7 @@ import {
 import { CleanUpPolicy, Topic } from 'generated-sources';
 import { externalTopicPayload } from 'lib/fixtures/topics';
 import {
+  useClearTopicMessages,
   useDeleteTopic,
   useRecreateTopic,
   useTopicDetails,
@@ -31,9 +32,11 @@ jest.mock('lib/hooks/api/topics', () => ({
   useTopicDetails: jest.fn(),
   useDeleteTopic: jest.fn(),
   useRecreateTopic: jest.fn(),
+  useClearTopicMessages: jest.fn(),
 }));
 
 const unwrapMock = jest.fn();
+const clearTopicMessages = jest.fn();
 
 jest.mock('lib/hooks/redux', () => ({
   ...jest.requireActual('lib/hooks/redux'),
@@ -98,6 +101,9 @@ describe('Details', () => {
     (useRecreateTopic as jest.Mock).mockImplementation(() => ({
       mutateAsync: mockRecreate,
     }));
+    (useClearTopicMessages as jest.Mock).mockImplementation(() => ({
+      mutateAsync: clearTopicMessages,
+    }));
     (useAppDispatch as jest.Mock).mockImplementation(() => () => ({
       unwrap: unwrapMock,
     }));
@@ -145,7 +151,7 @@ describe('Details', () => {
           name: 'Confirm',
         })[0];
         await waitFor(() => userEvent.click(submitButton));
-        expect(unwrapMock).toHaveBeenCalledTimes(1);
+        expect(clearTopicMessages).toHaveBeenCalledTimes(1);
       });
 
       it('closes the modal when cancel button is clicked', async () => {

kafka-ui-react-app/src/lib/hooks/api/topics.ts

@@ -2,6 +2,7 @@ import {
   topicsApiClient as api,
   messagesApiClient as messagesApi,
   consumerGroupsApiClient,
+  messagesApiClient,
 } from 'lib/api';
 import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
 import {
@@ -233,6 +234,34 @@ export function useDeleteTopic(clusterName: ClusterName) {
     }
   );
 }
+
+export function useClearTopicMessages(
+  clusterName: ClusterName,
+  partitions?: number[]
+) {
+  const client = useQueryClient();
+  return useMutation(
+    async (topicName: Topic['name']) => {
+      await messagesApiClient.deleteTopicMessages({
+        clusterName,
+        partitions,
+        topicName,
+      });
+      return topicName;
+    },
+
+    {
+      onSuccess: (topicName) => {
+        showSuccessAlert({
+          id: `message-${topicName}-${clusterName}-${partitions}`,
+          message: `${topicName} messages have been successfully cleared!`,
+        });
+        client.invalidateQueries(topicKeys.all(clusterName));
+      },
+    }
+  );
+}
+
 export function useRecreateTopic(props: GetTopicDetailsRequest) {
   const client = useQueryClient();
   return useMutation(() => api.recreateTopic(props), {

kafka-ui-react-app/src/redux/reducers/topicMessages/__test__/reducer.spec.ts

@@ -1,6 +1,5 @@
 import reducer, {
   addTopicMessage,
-  clearTopicMessages,
   resetTopicMessages,
   updateTopicMessagesMeta,
   updateTopicMessagesPhase,
@@ -12,9 +11,6 @@ import {
   topicMessagesMetaPayload,
 } from './fixtures';
 
-const clusterName = 'local';
-const topicName = 'localTopic';
-
 describe('TopicMessages reducer', () => {
   it('Adds new message', () => {
     const state = reducer(
@@ -67,24 +63,6 @@ describe('TopicMessages reducer', () => {
     expect(newState.messages.length).toEqual(0);
   });
 
-  it('clear messages', () => {
-    const state = reducer(
-      undefined,
-      addTopicMessage({ message: topicMessagePayload })
-    );
-    expect(state.messages.length).toEqual(1);
-
-    expect(
-      reducer(state, {
-        type: clearTopicMessages.fulfilled,
-        payload: { clusterName, topicName },
-      })
-    ).toEqual({
-      ...state,
-      messages: [],
-    });
-  });
-
   it('Updates Topic Messages Phase', () => {
     const phase = 'Polling';
 

kafka-ui-react-app/src/redux/reducers/topicMessages/topicMessagesSlice.ts

@@ -1,36 +1,6 @@
-import { createAsyncThunk, createSlice } from '@reduxjs/toolkit';
-import { TopicMessagesState, ClusterName, TopicName } from 'redux/interfaces';
+import { createSlice } from '@reduxjs/toolkit';
+import { TopicMessagesState } from 'redux/interfaces';
 import { TopicMessage } from 'generated-sources';
-import {
-  getResponse,
-  showServerError,
-  showSuccessAlert,
-} from 'lib/errorHandling';
-import { messagesApiClient } from 'lib/api';
-
-export const clearTopicMessages = createAsyncThunk<
-  undefined,
-  { clusterName: ClusterName; topicName: TopicName; partitions?: number[] }
->(
-  'topicMessages/clearTopicMessages',
-  async ({ clusterName, topicName, partitions }, { rejectWithValue }) => {
-    try {
-      await messagesApiClient.deleteTopicMessages({
-        clusterName,
-        topicName,
-        partitions,
-      });
-      showSuccessAlert({
-        id: `message-${topicName}-${clusterName}-${partitions}`,
-        message: `${topicName} messages have been successfully cleared!`,
-      });
-      return undefined;
-    } catch (err) {
-      showServerError(err as Response);
-      return rejectWithValue(await getResponse(err as Response));
-    }
-  }
-);
 
 export const initialState: TopicMessagesState = {
   messages: [],
@@ -68,11 +38,6 @@ const topicMessagesSlice = createSlice({
       state.isFetching = action.payload;
     },
   },
-  extraReducers: (builder) => {
-    builder.addCase(clearTopicMessages.fulfilled, (state) => {
-      state.messages = [];
-    });
-  },
 });
 
 export const {

pom.xml

@@ -43,7 +43,7 @@
         <spring-security.version>5.7.5</spring-security.version>
         <kafka-ui-serde-api.version>1.0.0</kafka-ui-serde-api.version>
         <odd-oddrn-generator.version>0.1.15</odd-oddrn-generator.version>
-        <odd-oddrn-client.version>0.1.19</odd-oddrn-client.version>
+        <odd-oddrn-client.version>0.1.23</odd-oddrn-client.version>
 
         <!-- Test dependency versions -->
         <junit.version>5.9.1</junit.version>