[issues-211] - Clearing up messages from a topic (#378)

Co-authored-by: mbovtryuk <mbovtryuk@provectus.com>

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

@@ -21,6 +21,11 @@ interface Props {
   totalPages: number;
   fetchTopicsList(props: FetchTopicsListParams): void;
   deleteTopic(topicName: TopicName, clusterName: ClusterName): void;
+  clearTopicMessages(
+    topicName: TopicName,
+    clusterName: ClusterName,
+    partitions?: number[]
+  ): void;
 }
 
 const List: React.FC<Props> = ({
@@ -30,6 +35,7 @@ const List: React.FC<Props> = ({
   totalPages,
   fetchTopicsList,
   deleteTopic,
+  clearTopicMessages,
 }) => {
   const { isReadOnly } = React.useContext(ClusterContext);
   const { clusterName } = useParams<{ clusterName: ClusterName }>();
@@ -99,6 +105,7 @@ const List: React.FC<Props> = ({
                     key={topic.name}
                     topic={topic}
                     deleteTopic={deleteTopic}
+                    clearTopicMessages={clearTopicMessages}
                   />
                 ))}
                 {items.length === 0 && (

kafka-ui-react-app/src/components/Topics/List/ListContainer.ts

@@ -1,6 +1,10 @@
 import { connect } from 'react-redux';
 import { RootState } from 'redux/interfaces';
-import { fetchTopicsList, deleteTopic } from 'redux/actions';
+import {
+  fetchTopicsList,
+  deleteTopic,
+  clearTopicMessages,
+} from 'redux/actions';
 import {
   getTopicList,
   getExternalTopicList,
@@ -19,6 +23,7 @@ const mapStateToProps = (state: RootState) => ({
 const mapDispatchToProps = {
   fetchTopicsList,
   deleteTopic,
+  clearTopicMessages,
 };
 
 export default connect(mapStateToProps, mapDispatchToProps)(List);

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

@@ -14,12 +14,14 @@ export interface ListItemProps {
   topic: TopicWithDetailedInfo;
   deleteTopic: (clusterName: ClusterName, topicName: TopicName) => void;
   clusterName: ClusterName;
+  clearTopicMessages(topicName: TopicName, clusterName: ClusterName): void;
 }
 
 const ListItem: React.FC<ListItemProps> = ({
   topic: { name, internal, partitions },
   deleteTopic,
   clusterName,
+  clearTopicMessages,
 }) => {
   const [
     isDeleteTopicConfirmationVisible,
@@ -41,6 +43,10 @@ const ListItem: React.FC<ListItemProps> = ({
     deleteTopic(clusterName, name);
   }, [clusterName, name]);
 
+  const clearTopicMessagesHandler = React.useCallback(() => {
+    clearTopicMessages(clusterName, name);
+  }, [clusterName, name]);
+
   return (
     <tr>
       <td className="has-text-overflow-ellipsis">
@@ -70,6 +76,9 @@ const ListItem: React.FC<ListItemProps> = ({
             }
             right
           >
+            <DropdownItem onClick={clearTopicMessagesHandler}>
+              <span className="has-text-danger">Clear Messages</span>
+            </DropdownItem>
             <DropdownItem
               onClick={() => setDeleteTopicConfirmationVisible(true)}
             >

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

@@ -23,6 +23,7 @@ describe('List', () => {
               totalPages={1}
               fetchTopicsList={jest.fn()}
               deleteTopic={jest.fn()}
+              clearTopicMessages={jest.fn()}
             />
           </ClusterContext.Provider>
         </StaticRouter>
@@ -49,6 +50,7 @@ describe('List', () => {
               totalPages={1}
               fetchTopicsList={jest.fn()}
               deleteTopic={jest.fn()}
+              clearTopicMessages={jest.fn()}
             />
           </ClusterContext.Provider>
         </StaticRouter>

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

@@ -9,6 +9,7 @@ import ListItem, { ListItemProps } from '../ListItem';
 
 const mockDelete = jest.fn();
 const clusterName = 'local';
+const mockDeleteMessages = jest.fn();
 
 jest.mock(
   'components/common/ConfirmationModal/ConfirmationModal',
@@ -21,14 +22,25 @@ describe('ListItem', () => {
       topic={internalTopicPayload}
       deleteTopic={mockDelete}
       clusterName={clusterName}
+      clearTopicMessages={mockDeleteMessages}
       {...props}
     />
   );
 
+  it('triggers the deleting messages when clicked on the delete messages button', () => {
+    const component = shallow(setupComponent());
+    component.find('DropdownItem').at(0).simulate('click');
+    expect(mockDeleteMessages).toBeCalledTimes(1);
+    expect(mockDeleteMessages).toBeCalledWith(
+      clusterName,
+      internalTopicPayload.name
+    );
+  });
+
   it('triggers the deleteTopic when clicked on the delete button', () => {
     const wrapper = shallow(setupComponent());
     expect(wrapper.find('mock-ConfirmationModal').prop('isOpen')).toBeFalsy();
-    wrapper.find('DropdownItem').last().simulate('click');
+    wrapper.find('DropdownItem').at(1).simulate('click');
     const modal = wrapper.find('mock-ConfirmationModal');
     expect(modal.prop('isOpen')).toBeTruthy();
     modal.simulate('confirm');

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

@@ -19,9 +19,15 @@ interface Props extends Topic, TopicDetails {
   clusterName: ClusterName;
   topicName: TopicName;
   deleteTopic: (clusterName: ClusterName, topicName: TopicName) => void;
+  clearTopicMessages(clusterName: ClusterName, topicName: TopicName): void;
 }
 
-const Details: React.FC<Props> = ({ clusterName, topicName, deleteTopic }) => {
+const Details: React.FC<Props> = ({
+  clusterName,
+  topicName,
+  deleteTopic,
+  clearTopicMessages,
+}) => {
   const history = useHistory();
   const { isReadOnly } = React.useContext(ClusterContext);
   const [
@@ -33,6 +39,10 @@ const Details: React.FC<Props> = ({ clusterName, topicName, deleteTopic }) => {
     history.push(clusterTopicsPath(clusterName));
   }, [clusterName, topicName]);
 
+  const clearTopicMessagesHandler = React.useCallback(() => {
+    clearTopicMessages(clusterName, topicName);
+  }, [clusterName, topicName]);
+
   return (
     <div className="box">
       <nav className="navbar" role="navigation">
@@ -66,6 +76,13 @@ const Details: React.FC<Props> = ({ clusterName, topicName, deleteTopic }) => {
           <div className="buttons">
             {!isReadOnly && (
               <>
+                <button
+                  type="button"
+                  className="button is-danger"
+                  onClick={clearTopicMessagesHandler}
+                >
+                  Clear All Messages
+                </button>
                 <button
                   className="button is-danger"
                   type="button"

kafka-ui-react-app/src/components/Topics/Topic/Details/DetailsContainer.ts

@@ -1,7 +1,8 @@
 import { connect } from 'react-redux';
 import { ClusterName, RootState, TopicName } from 'redux/interfaces';
 import { withRouter, RouteComponentProps } from 'react-router-dom';
-import { deleteTopic } from 'redux/actions';
+import { deleteTopic, clearTopicMessages } from 'redux/actions';
+
 import Details from './Details';
 
 interface RouteProps {
@@ -25,6 +26,7 @@ const mapStateToProps = (
 
 const mapDispatchToProps = {
   deleteTopic,
+  clearTopicMessages,
 };
 
 export default withRouter(

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

@@ -1,10 +1,21 @@
 import React from 'react';
 import { Topic, TopicDetails } from 'generated-sources';
+import { ClusterName, TopicName } from 'redux/interfaces';
+import Dropdown from 'components/common/Dropdown/Dropdown';
+import DropdownItem from 'components/common/Dropdown/DropdownItem';
 import MetricsWrapper from 'components/common/Dashboard/MetricsWrapper';
 import Indicator from 'components/common/Dashboard/Indicator';
 import BytesFormatted from 'components/common/BytesFormatted/BytesFormatted';
 
-interface Props extends Topic, TopicDetails {}
+interface Props extends Topic, TopicDetails {
+  clusterName: ClusterName;
+  topicName: TopicName;
+  clearTopicMessages(
+    clusterName: ClusterName,
+    topicName: TopicName,
+    partitions?: number[]
+  ): void;
+}
 
 const Overview: React.FC<Props> = ({
   partitions,
@@ -16,6 +27,9 @@ const Overview: React.FC<Props> = ({
   replicationFactor,
   segmentSize,
   segmentCount,
+  clusterName,
+  topicName,
+  clearTopicMessages,
 }) => (
   <>
     <MetricsWrapper>
@@ -43,28 +57,45 @@ const Overview: React.FC<Props> = ({
       <Indicator label="Segment count">{segmentCount}</Indicator>
     </MetricsWrapper>
     <div className="box">
-      <div className="table-container">
-        <table className="table is-striped is-fullwidth">
-          <thead>
-            <tr>
-              <th>Partition ID</th>
-              <th>Broker leader</th>
-              <th>Min offset</th>
-              <th>Max offset</th>
+      <table className="table is-striped is-fullwidth">
+        <thead>
+          <tr>
+            <th>Partition ID</th>
+            <th>Broker leader</th>
+            <th>Min offset</th>
+            <th>Max offset</th>
+            <th> </th>
+          </tr>
+        </thead>
+        <tbody>
+          {partitions?.map(({ partition, leader, offsetMin, offsetMax }) => (
+            <tr key={`partition-list-item-key-${partition}`}>
+              <td>{partition}</td>
+              <td>{leader}</td>
+              <td>{offsetMin}</td>
+              <td>{offsetMax}</td>
+              <td className="has-text-right">
+                <Dropdown
+                  label={
+                    <span className="icon">
+                      <i className="fas fa-cog" />
+                    </span>
+                  }
+                  right
+                >
+                  <DropdownItem
+                    onClick={() =>
+                      clearTopicMessages(clusterName, topicName, [partition])
+                    }
+                  >
+                    <span className="has-text-danger">Clear Messages</span>
+                  </DropdownItem>
+                </Dropdown>
+              </td>
             </tr>
-          </thead>
-          <tbody>
-            {partitions?.map(({ partition, leader, offsetMin, offsetMax }) => (
-              <tr key={`partition-list-item-key-${partition}`}>
-                <td>{partition}</td>
-                <td>{leader}</td>
-                <td>{offsetMin}</td>
-                <td>{offsetMax}</td>
-              </tr>
-            ))}
-          </tbody>
-        </table>
-      </div>
+          ))}
+        </tbody>
+      </table>
     </div>
   </>
 );

kafka-ui-react-app/src/components/Topics/Topic/Details/Overview/OverviewContainer.ts

@@ -2,6 +2,7 @@ import { connect } from 'react-redux';
 import { RootState, TopicName, ClusterName } from 'redux/interfaces';
 import { getTopicByName } from 'redux/reducers/topics/selectors';
 import { withRouter, RouteComponentProps } from 'react-router-dom';
+import { clearTopicMessages } from 'redux/actions';
 import Overview from './Overview';
 
 interface RouteProps {
@@ -15,11 +16,19 @@ const mapStateToProps = (
   state: RootState,
   {
     match: {
-      params: { topicName },
+      params: { topicName, clusterName },
     },
   }: OwnProps
 ) => ({
   ...getTopicByName(state, topicName),
+  topicName,
+  clusterName,
 });
 
-export default withRouter(connect(mapStateToProps)(Overview));
+const mapDispatchToProps = {
+  clearTopicMessages,
+};
+
+export default withRouter(
+  connect(mapStateToProps, mapDispatchToProps)(Overview)
+);

kafka-ui-react-app/src/redux/actions/__test__/actions.spec.ts

@@ -109,4 +109,26 @@ describe('Actions', () => {
       });
     });
   });
+
+  describe('clearMessagesTopicAction', () => {
+    it('creates a REQUEST action', () => {
+      expect(actions.clearMessagesTopicAction.request()).toEqual({
+        type: 'CLEAR_TOPIC_MESSAGES__REQUEST',
+      });
+    });
+
+    it('creates a SUCCESS action', () => {
+      expect(actions.clearMessagesTopicAction.success('topic')).toEqual({
+        type: 'CLEAR_TOPIC_MESSAGES__SUCCESS',
+        payload: 'topic',
+      });
+    });
+
+    it('creates a FAILURE action', () => {
+      expect(actions.clearMessagesTopicAction.failure({})).toEqual({
+        type: 'CLEAR_TOPIC_MESSAGES__FAILURE',
+        payload: {},
+      });
+    });
+  });
 });

kafka-ui-react-app/src/redux/actions/__test__/thunks/topics.spec.ts

@@ -43,4 +43,54 @@ describe('Thunks', () => {
       }
     });
   });
+
+  describe('clearTopicMessages', () => {
+    it('creates CLEAR_TOPIC_MESSAGES__SUCCESS when deleting existing messages', async () => {
+      fetchMock.deleteOnce(
+        `/api/clusters/${clusterName}/topics/${topicName}/messages`,
+        200
+      );
+      await store.dispatch(thunks.clearTopicMessages(clusterName, topicName));
+      expect(store.getActions()).toEqual([
+        actions.clearMessagesTopicAction.request(),
+        actions.clearMessagesTopicAction.success(topicName),
+      ]);
+    });
+
+    it('creates CLEAR_TOPIC_MESSAGES__FAILURE when deleting existing messages', async () => {
+      fetchMock.deleteOnce(
+        `/api/clusters/${clusterName}/topics/${topicName}/messages`,
+        404
+      );
+      try {
+        await store.dispatch(thunks.clearTopicMessages(clusterName, topicName));
+      } catch (error) {
+        expect(error.status).toEqual(404);
+        expect(store.getActions()).toEqual([
+          actions.clearMessagesTopicAction.request(),
+          actions.clearMessagesTopicAction.failure({}),
+        ]);
+      }
+    });
+  });
+
+  describe('fetchTopicMessages', () => {
+    it('creates GET_TOPIC_MESSAGES__FAILURE when deleting existing messages', async () => {
+      fetchMock.getOnce(
+        `/api/clusters/${clusterName}/topics/${topicName}/messages`,
+        404
+      );
+      try {
+        await store.dispatch(
+          thunks.fetchTopicMessages(clusterName, topicName, {})
+        );
+      } catch (error) {
+        expect(error.status).toEqual(404);
+        expect(store.getActions()).toEqual([
+          actions.fetchTopicMessagesAction.request(),
+          actions.fetchTopicMessagesAction.failure(),
+        ]);
+      }
+    });
+  });
 });

kafka-ui-react-app/src/redux/actions/actions.ts

@@ -60,6 +60,12 @@ export const fetchTopicMessagesAction = createAsyncAction(
   'GET_TOPIC_MESSAGES__FAILURE'
 )<undefined, TopicMessage[], undefined>();
 
+export const clearMessagesTopicAction = createAsyncAction(
+  'CLEAR_TOPIC_MESSAGES__REQUEST',
+  'CLEAR_TOPIC_MESSAGES__SUCCESS',
+  'CLEAR_TOPIC_MESSAGES__FAILURE'
+)<undefined, TopicName, { alert?: FailurePayload }>();
+
 export const fetchTopicDetailsAction = createAsyncAction(
   'GET_TOPIC_DETAILS__REQUEST',
   'GET_TOPIC_DETAILS__SUCCESS',

kafka-ui-react-app/src/redux/actions/thunks/topics.ts

@@ -81,6 +81,30 @@ export const fetchTopicMessages = (
   }
 };
 
+export const clearTopicMessages = (
+  clusterName: ClusterName,
+  topicName: TopicName,
+  partitions?: number[]
+): PromiseThunkResult => async (dispatch) => {
+  dispatch(actions.clearMessagesTopicAction.request());
+  try {
+    await messagesApiClient.deleteTopicMessages({
+      clusterName,
+      topicName,
+      partitions,
+    });
+    dispatch(actions.clearMessagesTopicAction.success(topicName));
+  } catch (e) {
+    const response = await getResponse(e);
+    const alert: FailurePayload = {
+      subject: [clusterName, topicName, partitions].join('-'),
+      title: `Clear Topic Messages`,
+      response,
+    };
+    dispatch(actions.clearMessagesTopicAction.failure({ alert }));
+  }
+};
+
 export const fetchTopicDetails = (
   clusterName: ClusterName,
   topicName: TopicName

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

@@ -1,29 +1,33 @@
-import { deleteTopicAction } from 'redux/actions';
+import { deleteTopicAction, clearMessagesTopicAction } from 'redux/actions';
 import reducer from '../reducer';
 
+const topic = {
+  name: 'topic',
+  id: 'id',
+};
+
+const state = {
+  byName: {
+    [topic.name]: topic,
+  },
+  allNames: [topic.name],
+  messages: [],
+  totalPages: 1,
+};
+
 describe('topics reducer', () => {
   it('deletes the topic from the list on DELETE_TOPIC__SUCCESS', () => {
-    const topic = {
-      name: 'topic',
-      id: 'id',
-    };
-    expect(
-      reducer(
-        {
-          byName: {
-            [topic.name]: topic,
-          },
-          allNames: [topic.name],
-          messages: [],
-          totalPages: 1,
-        },
-        deleteTopicAction.success(topic.name)
-      )
-    ).toEqual({
+    expect(reducer(state, deleteTopicAction.success(topic.name))).toEqual({
       byName: {},
       allNames: [],
       messages: [],
       totalPages: 1,
     });
   });
+
+  it('delete topic messages on CLEAR_TOPIC_MESSAGES__SUCCESS', () => {
+    expect(
+      reducer(state, clearMessagesTopicAction.success(topic.name))
+    ).toEqual(state);
+  });
 });

kafka-ui-react-app/src/redux/reducers/topics/reducer.ts

@@ -53,6 +53,12 @@ const reducer = (state = initialState, action: Action): TopicsState => {
       );
       return newState;
     }
+    case getType(actions.clearMessagesTopicAction.success): {
+      return {
+        ...state,
+        messages: [],
+      };
+    }
     default:
       return state;
   }