diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
index e0b20d6c93f..82d505c8cf3 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java
@@ -45,6 +45,7 @@ public static class Cluster {
     MetricsConfigData metrics;
     Map<String, Object> properties;
     boolean readOnly = false;
+    boolean messageDownloadAllowed = false;
     List<SerdeConfig> serde;
     String defaultKeySerde;
     String defaultValueSerde;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterState.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterState.java
index 28e9a7413a3..9008a47de12 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterState.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalClusterState.java
@@ -27,6 +27,7 @@ public class InternalClusterState {
   private BigDecimal bytesInPerSec;
   private BigDecimal bytesOutPerSec;
   private Boolean readOnly;
+  private Boolean isMessageDownloadAllowed;
 
   public InternalClusterState(KafkaCluster cluster, Statistics statistics) {
     name = cluster.getName();
@@ -75,6 +76,7 @@ public InternalClusterState(KafkaCluster cluster, Statistics statistics) {
     outOfSyncReplicasCount = partitionsStats.getOutOfSyncReplicasCount();
     underReplicatedPartitionCount = partitionsStats.getUnderReplicatedPartitionCount();
     readOnly = cluster.isReadOnly();
+    isMessageDownloadAllowed = cluster.isMessageDownloadAllowed();
   }
 
 }
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
index 1e2903dbcc9..6771ab4bd1a 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/KafkaCluster.java
@@ -25,6 +25,7 @@ public class KafkaCluster {
   private final String bootstrapServers;
   private final Properties properties;
   private final boolean readOnly;
+  private final boolean isMessageDownloadAllowed;
   private final MetricsConfig metricsConfig;
   private final DataMasking masking;
   private final PollingSettings pollingSettings;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaClusterFactory.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaClusterFactory.java
index 964b25473d3..d4f849e29f5 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaClusterFactory.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/KafkaClusterFactory.java
@@ -54,6 +54,7 @@ public KafkaCluster create(ClustersProperties properties,
     builder.bootstrapServers(clusterProperties.getBootstrapServers());
     builder.properties(convertProperties(clusterProperties.getProperties()));
     builder.readOnly(clusterProperties.isReadOnly());
+    builder.isMessageDownloadAllowed(clusterProperties.isMessageDownloadAllowed());
     builder.masking(DataMasking.create(clusterProperties.getMasking()));
     builder.pollingSettings(PollingSettings.create(clusterProperties, properties));
 
diff --git a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
index ae51d31568f..dda5543f636 100644
--- a/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
+++ b/kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
@@ -2189,6 +2189,8 @@ components:
           type: number
         readOnly:
           type: boolean
+        isMessageDownloadAllowed:
+          type: boolean
         version:
           type: string
         features:
diff --git a/kafka-ui-react-app/src/components/ClusterPage/ClusterPage.tsx b/kafka-ui-react-app/src/components/ClusterPage/ClusterPage.tsx
index 29d2015f612..26bb75a8297 100644
--- a/kafka-ui-react-app/src/components/ClusterPage/ClusterPage.tsx
+++ b/kafka-ui-react-app/src/components/ClusterPage/ClusterPage.tsx
@@ -56,6 +56,7 @@ const ClusterPage: React.FC = () => {
       hasAclViewConfigured:
         features.includes(ClusterFeaturesEnum.KAFKA_ACL_VIEW) ||
         features.includes(ClusterFeaturesEnum.KAFKA_ACL_EDIT),
+      isMessageDownloadAllowed: cluster?.isMessageDownloadAllowed || false,
     };
   }, [clusterName, data]);
 
diff --git a/kafka-ui-react-app/src/components/Topics/Topic/Messages/MessagesTable.tsx b/kafka-ui-react-app/src/components/Topics/Topic/Messages/MessagesTable.tsx
index 10e4f4dfa17..b6f053c7f65 100644
--- a/kafka-ui-react-app/src/components/Topics/Topic/Messages/MessagesTable.tsx
+++ b/kafka-ui-react-app/src/components/Topics/Topic/Messages/MessagesTable.tsx
@@ -13,6 +13,7 @@ import { Button } from 'components/common/Button/Button';
 import { useSearchParams } from 'react-router-dom';
 import { MESSAGES_PER_PAGE } from 'lib/constants';
 import * as S from 'components/common/NewTable/Table.styled';
+import ClusterContext from 'components/contexts/ClusterContext';
 
 import PreviewModal from './PreviewModal';
 import Message, { PreviewFilter } from './Message';
@@ -28,6 +29,7 @@ const MessagesTable: React.FC = () => {
   const { isLive } = useContext(TopicMessagesContext);
 
   const messages = useAppSelector(getTopicMessges);
+  const { isMessageDownloadAllowed } = useContext(ClusterContext);
   const isFetching = useAppSelector(getIsTopicMessagesFetching);
 
   const isTailing = isLive && isFetching;
@@ -51,6 +53,28 @@ const MessagesTable: React.FC = () => {
     setSearchParams(searchParams);
   };
 
+  const handleDownload = () => {
+    const download = (filename: string, content: Blob) => {
+      // create anchor tag to download file
+      const url = URL.createObjectURL(content);
+      const link = document.createElement('a');
+      link.href = url;
+      link.setAttribute('download', filename);
+      document.body.appendChild(link);
+
+      // download file
+      link.click();
+
+      // clean up
+      document.body.removeChild(link);
+      URL.revokeObjectURL(url);
+    };
+
+    const jsonString = JSON.stringify(messages);
+    const content = new Blob([jsonString], { type: 'application/json' });
+    download('download.json', content);
+  };
+
   return (
     <div style={{ position: 'relative' }}>
       {previewFor !== null && (
@@ -136,6 +160,14 @@ const MessagesTable: React.FC = () => {
           >
             Next →
           </Button>
+          <Button
+            buttonType="secondary"
+            buttonSize="L"
+            onClick={handleDownload}
+            disabled={!isMessageDownloadAllowed}
+          >
+            Download
+          </Button>
         </S.Pages>
       </S.Pagination>
     </div>
diff --git a/kafka-ui-react-app/src/components/contexts/ClusterContext.ts b/kafka-ui-react-app/src/components/contexts/ClusterContext.ts
index 3c50fcf1443..4fabde6ecb3 100644
--- a/kafka-ui-react-app/src/components/contexts/ClusterContext.ts
+++ b/kafka-ui-react-app/src/components/contexts/ClusterContext.ts
@@ -5,6 +5,7 @@ export interface ContextProps {
   hasKafkaConnectConfigured: boolean;
   hasSchemaRegistryConfigured: boolean;
   isTopicDeletionAllowed: boolean;
+  isMessageDownloadAllowed: boolean;
 }
 
 export const initialValue: ContextProps = {
@@ -12,6 +13,7 @@ export const initialValue: ContextProps = {
   hasKafkaConnectConfigured: false,
   hasSchemaRegistryConfigured: false,
   isTopicDeletionAllowed: true,
+  isMessageDownloadAllowed: false,
 };
 const ClusterContext = React.createContext(initialValue);