Very similar to #998.  It is somewhat similar indeed, but #998 is about re-creating single topic to clear out all data and "reset" all offsets to 0-state. This request is more about making a "clone" of one topic in a simple way. Just clarifying this point to avoid confusion. Yeah I get it :) That's a note for us about implementation
I am not sure whether we need to restrict the number of cloned messages for a particular topic. 
For now, the limit Integer.MAX_VALUE(about 10^9 messages). Maybe it is worth removing it or making it customized? looks like loadTopic will return TopicNotFoundException not empty mono in case of non-existing topic, and we have GlobalErrorWebExceptionHandler that should catch and map exception to proper status. So, no need to handle it here. Let's clone just the topic with its configuration without messages.
Clone topic functionality
### Is your proposal related to a problem?

I often find myself creating a topic that has exactly same configuration as existing one. In which case I need to have 2 windows open side-by-side to see source configuration and replicate it in new topic. That is a bit troublesome.

### Describe the solution you'd like

1. In topic list view a button to clone selected topic (when single topic selected)
2. (optional) clone with data (i.e. spawn consumer reading from beginning of the source topic to the end and writing messages to destination topic)

### Describe alternatives you've considered

N/A. Currently copying existing topic configuration needs to be done manually

### Additional context

N/A

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml
kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaTopicCreateTests.java
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
index b05d19d8537..3ebe2b03a70 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java
@@ -46,6 +46,13 @@ public Mono<ResponseEntity<TopicDTO>> recreateTopic(String clusterName,
         .map(s -> new ResponseEntity<>(s, HttpStatus.CREATED));
   }
 
+  @Override
+  public Mono<ResponseEntity<TopicDTO>> cloneTopic(
+      String clusterName, String topicName, String newTopicName, ServerWebExchange exchange) {
+    return topicsService.cloneTopic(getCluster(clusterName), topicName, newTopicName)
+        .map(s -> new ResponseEntity<>(s, HttpStatus.CREATED));
+  }
+
   @Override
   public Mono<ResponseEntity<Void>> deleteTopic(
       String clusterName, String topicName, ServerWebExchange exchange) {
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
index 184a10e4823..b4629cf39c6 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java
@@ -428,6 +428,18 @@ public TopicMessageSchemaDTO getTopicSchema(KafkaCluster cluster, String topicNa
         .getTopicSchema(topicName);
   }
 
+  public Mono<TopicDTO> cloneTopic(
+      KafkaCluster cluster, String topicName, String newTopicName) {
+    return loadTopic(cluster, topicName).flatMap(topic ->
+        adminClientService.get(cluster).flatMap(ac -> ac.createTopic(newTopicName,
+            topic.getPartitionCount(),
+            (short) topic.getReplicationFactor(),
+            topic.getTopicConfigs()
+                .stream()
+                .collect(Collectors.toMap(InternalTopicConfig::getName, InternalTopicConfig::getValue)))
+        ).thenReturn(newTopicName).flatMap(a -> loadTopic(cluster, newTopicName)).map(clusterMapper::toTopic));
+  }
+
   @VisibleForTesting
   @lombok.Value
   static class Pagination {
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 8133a810d40..d4a5d3b36e3 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
@@ -331,6 +331,38 @@ paths:
               schema:
                 $ref: '#/components/schemas/Topic'
 
+  /api/clusters/{clusterName}/topics/{topicName}/clone:
+    post:
+      tags:
+        - Topics
+      summary: cloneTopic
+      operationId: cloneTopic
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: topicName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: newTopicName
+          in: query
+          required: true
+          schema:
+            type: string
+      responses:
+        201:
+          description: Created
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Topic'
+        404:
+          description: Not found
+
   /api/clusters/{clusterName}/topics/{topicName}:
     get:
       tags:

diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaTopicCreateTests.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaTopicCreateTests.java
index 558695556a8..3c47d7e9b51 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaTopicCreateTests.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/KafkaTopicCreateTests.java
@@ -76,4 +76,32 @@ void shouldRecreateExistingTopicSuccessfully() {
             .jsonPath("replicationFactor").isEqualTo(topicCreation.getReplicationFactor().toString())
             .jsonPath("name").isEqualTo(topicCreation.getName());
   }
+
+  @Test
+  void shouldCloneExistingTopicSuccessfully() {
+    TopicCreationDTO topicCreation = new TopicCreationDTO()
+        .replicationFactor(1)
+        .partitions(3)
+        .name(UUID.randomUUID().toString());
+    String clonedTopicName = UUID.randomUUID().toString();
+
+    webTestClient.post()
+        .uri("/api/clusters/{clusterName}/topics", LOCAL)
+        .bodyValue(topicCreation)
+        .exchange()
+        .expectStatus()
+        .isOk();
+
+    webTestClient.post()
+        .uri("/api/clusters/{clusterName}/topics/{topicName}/clone?newTopicName=" + clonedTopicName,
+            LOCAL, topicCreation.getName())
+        .exchange()
+        .expectStatus()
+        .isCreated()
+        .expectBody()
+        .jsonPath("partitionCount").isEqualTo(topicCreation.getPartitions().toString())
+        .jsonPath("replicationFactor").isEqualTo(topicCreation.getReplicationFactor().toString())
+        .jsonPath("name").isEqualTo(clonedTopicName);
+  }
+
 }
