Merge branch 'master' into issue-200-update-schema-subject-object

CODE-OF-CONDUCT.md

@@ -0,0 +1,132 @@
+
+# Contributor Covenant Code of Conduct
+
+## Our Pledge
+
+We as members, contributors, and leaders pledge to make participation in our
+community a harassment-free experience for everyone, regardless of age, body
+size, visible or invisible disability, ethnicity, sex characteristics, gender
+identity and expression, level of experience, education, socio-economic status,
+nationality, personal appearance, race, caste, color, religion, or sexual identity
+and orientation.
+
+We pledge to act and interact in ways that contribute to an open, welcoming,
+diverse, inclusive, and healthy community.
+
+## Our Standards
+
+Examples of behavior that contributes to a positive environment for our
+community include:
+
+* Demonstrating empathy and kindness toward other people
+* Being respectful of differing opinions, viewpoints, and experiences
+* Giving and gracefully accepting constructive feedback
+* Accepting responsibility and apologizing to those affected by our mistakes,
+  and learning from the experience
+* Focusing on what is best not just for us as individuals, but for the
+  overall community
+
+Examples of unacceptable behavior include:
+
+* The use of sexualized language or imagery, and sexual attention or
+  advances of any kind
+* Trolling, insulting or derogatory comments, and personal or political attacks
+* Public or private harassment
+* Publishing others' private information, such as a physical or email
+  address, without their explicit permission
+* Other conduct which could reasonably be considered inappropriate in a
+  professional setting
+
+## Enforcement Responsibilities
+
+Community leaders are responsible for clarifying and enforcing our standards of
+acceptable behavior and will take appropriate and fair corrective action in
+response to any behavior that they deem inappropriate, threatening, offensive,
+or harmful.
+
+Community leaders have the right and responsibility to remove, edit, or reject
+comments, commits, code, wiki edits, issues, and other contributions that are
+not aligned to this Code of Conduct, and will communicate reasons for moderation
+decisions when appropriate.
+
+## Scope
+
+This Code of Conduct applies within all community spaces, and also applies when
+an individual is officially representing the community in public spaces.
+Examples of representing our community include using an official e-mail address,
+posting via an official social media account, or acting as an appointed
+representative at an online or offline event.
+
+## Enforcement
+
+Instances of abusive, harassing, or otherwise unacceptable behavior may be
+reported to the community leaders responsible for enforcement at email kafkaui@provectus.com.
+All complaints will be reviewed and investigated promptly and fairly.
+
+All community leaders are obligated to respect the privacy and security of the
+reporter of any incident.
+
+## Enforcement Guidelines
+
+Community leaders will follow these Community Impact Guidelines in determining
+the consequences for any action they deem in violation of this Code of Conduct:
+
+### 1. Correction
+
+**Community Impact**: Use of inappropriate language or other behavior deemed
+unprofessional or unwelcome in the community.
+
+**Consequence**: A private, written warning from community leaders, providing
+clarity around the nature of the violation and an explanation of why the
+behavior was inappropriate. A public apology may be requested.
+
+### 2. Warning
+
+**Community Impact**: A violation through a single incident or series
+of actions.
+
+**Consequence**: A warning with consequences for continued behavior. No
+interaction with the people involved, including unsolicited interaction with
+those enforcing the Code of Conduct, for a specified period of time. This
+includes avoiding interactions in community spaces as well as external channels
+like social media. Violating these terms may lead to a temporary or
+permanent ban.
+
+### 3. Temporary Ban
+
+**Community Impact**: A serious violation of community standards, including
+sustained inappropriate behavior.
+
+**Consequence**: A temporary ban from any sort of interaction or public
+communication with the community for a specified period of time. No public or
+private interaction with the people involved, including unsolicited interaction
+with those enforcing the Code of Conduct, is allowed during this period.
+Violating these terms may lead to a permanent ban.
+
+### 4. Permanent Ban
+
+**Community Impact**: Demonstrating a pattern of violation of community
+standards, including sustained inappropriate behavior,  harassment of an
+individual, or aggression toward or disparagement of classes of individuals.
+
+**Consequence**: A permanent ban from any sort of public interaction within
+the community.
+
+## Attribution
+
+This Code of Conduct is adapted from the [Contributor Covenant][homepage],
+version 2.0, available at
+[https://www.contributor-covenant.org/version/2/0/code_of_conduct.html][v2.0].
+
+Community Impact Guidelines were inspired by 
+[Mozilla's code of conduct enforcement ladder][Mozilla CoC].
+
+For answers to common questions about this code of conduct, see the FAQ at
+[https://www.contributor-covenant.org/faq][FAQ]. Translations are available 
+at [https://www.contributor-covenant.org/translations][translations].
+
+[homepage]: https://www.contributor-covenant.org
+[v2.0]: https://www.contributor-covenant.org/version/2/0/code_of_conduct.html
+[Mozilla CoC]: https://github.com/mozilla/diversity
+[FAQ]: https://www.contributor-covenant.org/faq
+[translations]: https://www.contributor-covenant.org/translations

CONTRIBUTING.md

@@ -0,0 +1,16 @@
+# Contributing
+
+When contributing to this repository, please first discuss the change you wish to make via issue,
+email, or any other method with the owners of this repository before making a change. 
+
+Please note we have a code of conduct, please follow it in all your interactions with the project.
+
+## Pull Request Process
+
+1. Ensure any install or build dependencies are removed before the end of the layer when doing a 
+   build.
+2. Update the README.md with details of changes to the interface, this includes new environment 
+   variables, exposed ports, useful file locations and container parameters.
+3. Start Pull Request name with issue number (ex. #123)   
+4. You may merge the Pull Request in once you have the sign-off of two other developers, or if you 
+   do not have permission to do that, you may request the second reviewer to merge it for you.

LICENSE

@@ -0,0 +1,202 @@
+
+                                 Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/
+
+   TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION
+
+   1. Definitions.
+
+      "License" shall mean the terms and conditions for use, reproduction,
+      and distribution as defined by Sections 1 through 9 of this document.
+
+      "Licensor" shall mean the copyright owner or entity authorized by
+      the copyright owner that is granting the License.
+
+      "Legal Entity" shall mean the union of the acting entity and all
+      other entities that control, are controlled by, or are under common
+      control with that entity. For the purposes of this definition,
+      "control" means (i) the power, direct or indirect, to cause the
+      direction or management of such entity, whether by contract or
+      otherwise, or (ii) ownership of fifty percent (50%) or more of the
+      outstanding shares, or (iii) beneficial ownership of such entity.
+
+      "You" (or "Your") shall mean an individual or Legal Entity
+      exercising permissions granted by this License.
+
+      "Source" form shall mean the preferred form for making modifications,
+      including but not limited to software source code, documentation
+      source, and configuration files.
+
+      "Object" form shall mean any form resulting from mechanical
+      transformation or translation of a Source form, including but
+      not limited to compiled object code, generated documentation,
+      and conversions to other media types.
+
+      "Work" shall mean the work of authorship, whether in Source or
+      Object form, made available under the License, as indicated by a
+      copyright notice that is included in or attached to the work
+      (an example is provided in the Appendix below).
+
+      "Derivative Works" shall mean any work, whether in Source or Object
+      form, that is based on (or derived from) the Work and for which the
+      editorial revisions, annotations, elaborations, or other modifications
+      represent, as a whole, an original work of authorship. For the purposes
+      of this License, Derivative Works shall not include works that remain
+      separable from, or merely link (or bind by name) to the interfaces of,
+      the Work and Derivative Works thereof.
+
+      "Contribution" shall mean any work of authorship, including
+      the original version of the Work and any modifications or additions
+      to that Work or Derivative Works thereof, that is intentionally
+      submitted to Licensor for inclusion in the Work by the copyright owner
+      or by an individual or Legal Entity authorized to submit on behalf of
+      the copyright owner. For the purposes of this definition, "submitted"
+      means any form of electronic, verbal, or written communication sent
+      to the Licensor or its representatives, including but not limited to
+      communication on electronic mailing lists, source code control systems,
+      and issue tracking systems that are managed by, or on behalf of, the
+      Licensor for the purpose of discussing and improving the Work, but
+      excluding communication that is conspicuously marked or otherwise
+      designated in writing by the copyright owner as "Not a Contribution."
+
+      "Contributor" shall mean Licensor and any individual or Legal Entity
+      on behalf of whom a Contribution has been received by Licensor and
+      subsequently incorporated within the Work.
+
+   2. Grant of Copyright License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      copyright license to reproduce, prepare Derivative Works of,
+      publicly display, publicly perform, sublicense, and distribute the
+      Work and such Derivative Works in Source or Object form.
+
+   3. Grant of Patent License. Subject to the terms and conditions of
+      this License, each Contributor hereby grants to You a perpetual,
+      worldwide, non-exclusive, no-charge, royalty-free, irrevocable
+      (except as stated in this section) patent license to make, have made,
+      use, offer to sell, sell, import, and otherwise transfer the Work,
+      where such license applies only to those patent claims licensable
+      by such Contributor that are necessarily infringed by their
+      Contribution(s) alone or by combination of their Contribution(s)
+      with the Work to which such Contribution(s) was submitted. If You
+      institute patent litigation against any entity (including a
+      cross-claim or counterclaim in a lawsuit) alleging that the Work
+      or a Contribution incorporated within the Work constitutes direct
+      or contributory patent infringement, then any patent licenses
+      granted to You under this License for that Work shall terminate
+      as of the date such litigation is filed.
+
+   4. Redistribution. You may reproduce and distribute copies of the
+      Work or Derivative Works thereof in any medium, with or without
+      modifications, and in Source or Object form, provided that You
+      meet the following conditions:
+
+      (a) You must give any other recipients of the Work or
+          Derivative Works a copy of this License; and
+
+      (b) You must cause any modified files to carry prominent notices
+          stating that You changed the files; and
+
+      (c) You must retain, in the Source form of any Derivative Works
+          that You distribute, all copyright, patent, trademark, and
+          attribution notices from the Source form of the Work,
+          excluding those notices that do not pertain to any part of
+          the Derivative Works; and
+
+      (d) If the Work includes a "NOTICE" text file as part of its
+          distribution, then any Derivative Works that You distribute must
+          include a readable copy of the attribution notices contained
+          within such NOTICE file, excluding those notices that do not
+          pertain to any part of the Derivative Works, in at least one
+          of the following places: within a NOTICE text file distributed
+          as part of the Derivative Works; within the Source form or
+          documentation, if provided along with the Derivative Works; or,
+          within a display generated by the Derivative Works, if and
+          wherever such third-party notices normally appear. The contents
+          of the NOTICE file are for informational purposes only and
+          do not modify the License. You may add Your own attribution
+          notices within Derivative Works that You distribute, alongside
+          or as an addendum to the NOTICE text from the Work, provided
+          that such additional attribution notices cannot be construed
+          as modifying the License.
+
+      You may add Your own copyright statement to Your modifications and
+      may provide additional or different license terms and conditions
+      for use, reproduction, or distribution of Your modifications, or
+      for any such Derivative Works as a whole, provided Your use,
+      reproduction, and distribution of the Work otherwise complies with
+      the conditions stated in this License.
+
+   5. Submission of Contributions. Unless You explicitly state otherwise,
+      any Contribution intentionally submitted for inclusion in the Work
+      by You to the Licensor shall be under the terms and conditions of
+      this License, without any additional terms or conditions.
+      Notwithstanding the above, nothing herein shall supersede or modify
+      the terms of any separate license agreement you may have executed
+      with Licensor regarding such Contributions.
+
+   6. Trademarks. This License does not grant permission to use the trade
+      names, trademarks, service marks, or product names of the Licensor,
+      except as required for reasonable and customary use in describing the
+      origin of the Work and reproducing the content of the NOTICE file.
+
+   7. Disclaimer of Warranty. Unless required by applicable law or
+      agreed to in writing, Licensor provides the Work (and each
+      Contributor provides its Contributions) on an "AS IS" BASIS,
+      WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+      implied, including, without limitation, any warranties or conditions
+      of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A
+      PARTICULAR PURPOSE. You are solely responsible for determining the
+      appropriateness of using or redistributing the Work and assume any
+      risks associated with Your exercise of permissions under this License.
+
+   8. Limitation of Liability. In no event and under no legal theory,
+      whether in tort (including negligence), contract, or otherwise,
+      unless required by applicable law (such as deliberate and grossly
+      negligent acts) or agreed to in writing, shall any Contributor be
+      liable to You for damages, including any direct, indirect, special,
+      incidental, or consequential damages of any character arising as a
+      result of this License or out of the use or inability to use the
+      Work (including but not limited to damages for loss of goodwill,
+      work stoppage, computer failure or malfunction, or any and all
+      other commercial damages or losses), even if such Contributor
+      has been advised of the possibility of such damages.
+
+   9. Accepting Warranty or Additional Liability. While redistributing
+      the Work or Derivative Works thereof, You may choose to offer,
+      and charge a fee for, acceptance of support, warranty, indemnity,
+      or other liability obligations and/or rights consistent with this
+      License. However, in accepting such obligations, You may act only
+      on Your own behalf and on Your sole responsibility, not on behalf
+      of any other Contributor, and only if You agree to indemnify,
+      defend, and hold each Contributor harmless for any liability
+      incurred by, or claims asserted against, such Contributor by reason
+      of your accepting any such warranty or additional liability.
+
+   END OF TERMS AND CONDITIONS
+
+   APPENDIX: How to apply the Apache License to your work.
+
+      To apply the Apache License to your work, attach the following
+      boilerplate notice, with the fields enclosed by brackets "[]"
+      replaced with your own identifying information. (Don't include
+      the brackets!)  The text should be enclosed in the appropriate
+      comment syntax for the file format. We also recommend that a
+      file or class name and description of purpose be included on the
+      same "printed page" as the copyright notice for easier
+      identification within third-party archives.
+
+   Copyright 2020 CloudHut
+
+   Licensed under the Apache License, Version 2.0 (the "License");
+   you may not use this file except in compliance with the License.
+   You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+   Unless required by applicable law or agreed to in writing, software
+   distributed under the License is distributed on an "AS IS" BASIS,
+   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+   See the License for the specific language governing permissions and
+   limitations under the License.

docker/kafka-clusters-only.yaml

@@ -3,7 +3,7 @@ version: '2'
 services:
 
   zookeeper0:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.2.4
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
@@ -11,7 +11,7 @@ services:
     - 2181:2181
 
   kafka0:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     depends_on:
       - zookeeper0
     environment:
@@ -28,7 +28,7 @@ services:
       - 9997:9997
 
   kafka01:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     depends_on:
       - zookeeper0
     environment:
@@ -45,7 +45,7 @@ services:
       - 9999:9999
 
   zookeeper1:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.2.4
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
@@ -53,7 +53,7 @@ services:
       - 2182:2181
 
   kafka1:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     depends_on:
       - zookeeper1
     environment:
@@ -70,7 +70,7 @@ services:
       - 9998:9998 
 
   schemaregistry0:
-    image: confluentinc/cp-schema-registry:5.1.0
+    image: confluentinc/cp-schema-registry:5.2.4
     depends_on:
       - zookeeper0
       - kafka0
@@ -89,7 +89,7 @@ services:
     - 8081:8081
 
   kafka-connect0:
-    image: confluentinc/cp-kafka-connect:5.1.0
+    image: confluentinc/cp-kafka-connect:5.2.4
     ports:
       - 8083:8083
     depends_on:
@@ -115,7 +115,7 @@ services:
 
 
   kafka-init-topics:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     volumes:
        - ./message.json:/data/message.json
     depends_on:

docker/kafka-ui.yaml

@@ -31,7 +31,7 @@ services:
       KAFKA_CLUSTERS_1_KAFKACONNECT_0_ADDRESS: http://kafka-connect0:8083
 
   zookeeper0:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.2.4
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
@@ -39,7 +39,7 @@ services:
       - 2181:2181
 
   kafka0:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     depends_on:
       - zookeeper0
     ports:
@@ -56,13 +56,13 @@ services:
       KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=kafka0 -Dcom.sun.management.jmxremote.rmi.port=9997
 
   zookeeper1:
-    image: confluentinc/cp-zookeeper:5.1.0
+    image: confluentinc/cp-zookeeper:5.2.4
     environment:
       ZOOKEEPER_CLIENT_PORT: 2181
       ZOOKEEPER_TICK_TIME: 2000
 
   kafka1:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     depends_on:
       - zookeeper1
     ports:
@@ -79,7 +79,7 @@ services:
       KAFKA_JMX_OPTS: -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Djava.rmi.server.hostname=kafka1 -Dcom.sun.management.jmxremote.rmi.port=9998
 
   schemaregistry0:
-    image: confluentinc/cp-schema-registry:5.1.0
+    image: confluentinc/cp-schema-registry:5.2.4
     ports:
       - 8085:8085
     depends_on:
@@ -97,7 +97,7 @@ services:
       SCHEMA_REGISTRY_KAFKASTORE_TOPIC: _schemas
 
   kafka-connect0:
-    image: confluentinc/cp-kafka-connect:5.1.0
+    image: confluentinc/cp-kafka-connect:5.2.4
     ports:
       - 8083:8083
     depends_on:
@@ -122,7 +122,7 @@ services:
       CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
 
   kafka-init-topics:
-    image: confluentinc/cp-kafka:5.1.0
+    image: confluentinc/cp-kafka:5.2.4
     volumes:
        - ./message.json:/data/message.json
     depends_on:

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializer.java

@@ -150,7 +150,7 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
 			byte[] bytes = AvroSchemaUtils.toJson(avroRecord);
 			return parseJson(bytes);
 		} else {
-			return new HashMap<String,Object>();
+			return Map.of();
 		}
 	}
 
@@ -162,12 +162,16 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
 			byte[] bytes = ProtobufSchemaUtils.toJson(message);
 			return parseJson(bytes);
 		} else {
-			return new HashMap<String,Object>();
+			return Map.of();
 		}
 	}
 
 	private Object parseJsonRecord(ConsumerRecord<Bytes, Bytes> record) throws IOException {
-		byte[] valueBytes = record.value().get();
+		var value = record.value();
+		if (value == null) {
+			return Map.of();
+		}
+		byte[] valueBytes = value.get();
 		return parseJson(valueBytes);
 	}
 
@@ -178,6 +182,9 @@ public class SchemaRegistryRecordDeserializer implements RecordDeserializer {
 
 	private Object parseStringRecord(ConsumerRecord<Bytes, Bytes> record) {
 		String topic = record.topic();
+		if (record.value() == null) {
+			return Map.of();
+		}
 		byte[] valueBytes = record.value().get();
 		return stringDeserializer.deserialize(topic, valueBytes);
 	}

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

@@ -1,5 +1,6 @@
 package com.provectus.kafka.ui.cluster.service;
 
+import com.provectus.kafka.ui.cluster.exception.NotFoundException;
 import com.provectus.kafka.ui.cluster.mapper.ClusterMapper;
 import com.provectus.kafka.ui.cluster.model.ClustersStorage;
 import com.provectus.kafka.ui.cluster.model.ConsumerPosition;
@@ -169,12 +170,27 @@ public class ClusterService {
         ).orElse(Mono.empty());
     }
 
+    public Mono<Void> deleteTopic(String clusterName, String topicName) {
+        var cluster = clustersStorage.getClusterByName(clusterName)
+                .orElseThrow(() -> new NotFoundException("No such cluster"));
+        getTopicDetails(clusterName, topicName)
+                .orElseThrow(() -> new NotFoundException("No such topic"));
+        return kafkaService.deleteTopic(cluster, topicName)
+                .doOnNext(t -> updateCluster(topicName, clusterName, cluster));
+    }
+
     private KafkaCluster updateCluster(InternalTopic topic, String clusterName, KafkaCluster cluster) {
         final KafkaCluster updatedCluster = kafkaService.getUpdatedCluster(cluster, topic);
         clustersStorage.setKafkaCluster(clusterName, updatedCluster);
         return updatedCluster;
     }
 
+    private KafkaCluster updateCluster(String topicToDelete, String clusterName, KafkaCluster cluster) {
+        final KafkaCluster updatedCluster = kafkaService.getUpdatedCluster(cluster, topicToDelete);
+        clustersStorage.setKafkaCluster(clusterName, updatedCluster);
+        return updatedCluster;
+    }
+
     public Flux<TopicMessage> getMessages(String clusterName, String topicName, ConsumerPosition consumerPosition, String query, Integer limit) {
         return clustersStorage.getClusterByName(clusterName)
                 .map(c -> consumingService.loadMessages(c, topicName, consumerPosition, query, limit))

kafka-ui-api/src/main/java/com/provectus/kafka/ui/kafka/KafkaService.java

@@ -56,6 +56,12 @@ public class KafkaService {
         return cluster.toBuilder().topics(topics).build();
     }
 
+    public KafkaCluster getUpdatedCluster(KafkaCluster cluster, String topicToDelete) {
+        final Map<String, InternalTopic> topics = new HashMap<>(cluster.getTopics());
+        topics.remove(topicToDelete);
+        return cluster.toBuilder().topics(topics).build();
+    }
+
     @SneakyThrows
     public Mono<KafkaCluster> getUpdatedCluster(KafkaCluster cluster) {
         return getOrCreateAdminClient(cluster)
@@ -184,6 +190,13 @@ public class KafkaService {
         return getOrCreateAdminClient(cluster).flatMap(ac -> createTopic(ac.getAdminClient(), topicFormData));
     }
 
+    public Mono<Void> deleteTopic(KafkaCluster cluster, String topicName) {
+        return getOrCreateAdminClient(cluster)
+                .map(ExtendedAdminClient::getAdminClient)
+                .map(adminClient -> adminClient.deleteTopics(List.of(topicName)))
+                .then();
+    }
+
     @SneakyThrows
     public Mono<InternalTopic> createTopic(AdminClient adminClient, Mono<TopicFormData> topicFormData) {
         return topicFormData.flatMap(
@@ -208,12 +221,13 @@ public class KafkaService {
     }
 
     public Mono<ExtendedAdminClient> createAdminClient(KafkaCluster kafkaCluster) {
-        Properties properties = new Properties();
-        properties.putAll(kafkaCluster.getProperties());
-        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster.getBootstrapServers());
-        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, clientTimeout);
-        AdminClient adminClient = AdminClient.create(properties);
-        return ExtendedAdminClient.extendedAdminClient(adminClient);
+        return Mono.fromSupplier(() -> {
+            Properties properties = new Properties();
+            properties.putAll(kafkaCluster.getProperties());
+            properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaCluster.getBootstrapServers());
+            properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, clientTimeout);
+            return AdminClient.create(properties);
+        }).flatMap(ExtendedAdminClient::extendedAdminClient);
     }
 
     @SneakyThrows

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/MetricsRestController.java

@@ -160,6 +160,11 @@ public class MetricsRestController implements ApiClustersApi {
         return clusterService.updateTopic(clusterId, topicName, topicFormData).map(ResponseEntity::ok);
     }
 
+    @Override
+    public Mono<ResponseEntity<Void>> deleteTopic(String clusterName, String topicName, ServerWebExchange exchange) {
+        return clusterService.deleteTopic(clusterName, topicName).map(ResponseEntity::ok);
+    }
+
     @Override
     public Mono<ResponseEntity<CompatibilityLevel>> getGlobalSchemaCompatibilityLevel(String clusterName, ServerWebExchange exchange) {
         return schemaRegistryService.getGlobalSchemaCompatibilityLevel(clusterName)

kafka-ui-api/src/main/resources/application-local.yml

@@ -2,7 +2,7 @@ kafka:
   clusters:
     -
       name: local
-      bootstrapServers: localhost:9093
+      bootstrapServers: localhost:9092
       zookeeper: localhost:2181
       schemaRegistry: http://localhost:8081
       kafkaConnect:

kafka-ui-api/src/test/java/com/provectus/kafka/ui/cluster/deserialization/SchemaRegistryRecordDeserializerTest.java

@@ -0,0 +1,34 @@
+package com.provectus.kafka.ui.cluster.deserialization;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.provectus.kafka.ui.cluster.model.KafkaCluster;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.utils.Bytes;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class SchemaRegistryRecordDeserializerTest {
+
+    private final SchemaRegistryRecordDeserializer deserializer = new SchemaRegistryRecordDeserializer(
+            KafkaCluster.builder()
+                    .schemaNameTemplate("%s-value")
+                    .build(),
+            new ObjectMapper()
+    );
+
+    @Test
+    public void shouldDeserializeStringValue() {
+        var value = "test";
+        var deserializedRecord = deserializer.deserialize(new ConsumerRecord<>("topic", 1, 0, Bytes.wrap("key".getBytes()), Bytes.wrap(value.getBytes())));
+        assertEquals(value, deserializedRecord);
+    }
+
+    @Test
+    public void shouldDeserializeNullValueRecordToEmptyMap() {
+        var deserializedRecord = deserializer.deserialize(new ConsumerRecord<>("topic", 1, 0, Bytes.wrap("key".getBytes()), null));
+        assertEquals(Map.of(), deserializedRecord);
+    }
+}

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -215,6 +215,27 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/Topic'
+    delete:
+      tags:
+        - /api/clusters
+      summary: deleteTopic
+      operationId: deleteTopic
+      parameters:
+        - name: clusterName
+          in: path
+          required: true
+          schema:
+            type: string
+        - name: topicName
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        200:
+          description: OK
+        404:
+          description: Not found
 
   /api/clusters/{clusterName}/topics/{topicName}/config:
     get:

kafka-ui-react-app/src/components/common/Breadcrumb/Breadcrumb.tsx

@@ -1,7 +1,7 @@
 import React from 'react';
 import { NavLink } from 'react-router-dom';
 
-interface Link {
+export interface Link {
   label: string;
   href: string;
 }

kafka-ui-react-app/src/components/common/Breadcrumb/__tests__/Breadcrumb.spec.tsx

@@ -0,0 +1,48 @@
+import { mount, shallow } from 'enzyme';
+import React from 'react';
+import { StaticRouter } from 'react-router-dom';
+import Breadcrumb, { Link } from '../Breadcrumb';
+
+describe('Breadcrumb component', () => {
+  const links: Link[] = [
+    {
+      label: 'link1',
+      href: 'link1href',
+    },
+    {
+      label: 'link2',
+      href: 'link2href',
+    },
+    {
+      label: 'link3',
+      href: 'link3href',
+    },
+  ];
+
+  const child = <div className="child" />;
+
+  const component = mount(
+    <StaticRouter>
+      <Breadcrumb links={links}>{child}</Breadcrumb>
+    </StaticRouter>
+  );
+
+  it('renders the list of links', () => {
+    component.find(`NavLink`).forEach((link, idx) => {
+      expect(link.prop('to')).toEqual(links[idx].href);
+      expect(link.contains(links[idx].label)).toBeTruthy();
+    });
+  });
+  it('renders the children', () => {
+    const list = component.find('ul').children();
+    expect(list.last().containsMatchingElement(child)).toBeTruthy();
+  });
+  it('matches the snapshot', () => {
+    const shallowComponent = shallow(
+      <StaticRouter>
+        <Breadcrumb links={links}>{child}</Breadcrumb>
+      </StaticRouter>
+    );
+    expect(shallowComponent).toMatchSnapshot();
+  });
+});

kafka-ui-react-app/src/components/common/Breadcrumb/__tests__/__snapshots__/Breadcrumb.spec.tsx.snap

@@ -0,0 +1,49 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`Breadcrumb component matches the snapshot 1`] = `
+<Router
+  history={
+    Object {
+      "action": "POP",
+      "block": [Function],
+      "createHref": [Function],
+      "go": [Function],
+      "goBack": [Function],
+      "goForward": [Function],
+      "listen": [Function],
+      "location": Object {
+        "hash": "",
+        "pathname": "/",
+        "search": "",
+        "state": undefined,
+      },
+      "push": [Function],
+      "replace": [Function],
+    }
+  }
+  staticContext={Object {}}
+>
+  <Breadcrumb
+    links={
+      Array [
+        Object {
+          "href": "link1href",
+          "label": "link1",
+        },
+        Object {
+          "href": "link2href",
+          "label": "link2",
+        },
+        Object {
+          "href": "link3href",
+          "label": "link3",
+        },
+      ]
+    }
+  >
+    <div
+      className="child"
+    />
+  </Breadcrumb>
+</Router>
+`;

kafka-ui-react-app/src/components/common/BytesFormatted/BytesFormatted.tsx

@@ -5,20 +5,22 @@ interface Props {
   precision?: number;
 }
 
-const BytesFormatted: React.FC<Props> = ({ value, precision = 0 }) => {
-  const formatedValue = React.useMemo(() => {
-    const bytes = typeof value === 'string' ? parseInt(value, 10) : value;
-
-    const sizes = ['Bytes', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB'];
-    if (!bytes || bytes === 0) return [0, sizes[0]];
+export const sizes = ['Bytes', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB', 'YB'];
 
-    if (bytes < 1024) return [Math.ceil(bytes), sizes[0]];
-
-    const pow = Math.floor(Math.log2(bytes) / 10);
-    const multiplier = 10 ** (precision || 2);
-    return (
-      Math.round((bytes * multiplier) / 1024 ** pow) / multiplier + sizes[pow]
-    );
+const BytesFormatted: React.FC<Props> = ({ value, precision = 0 }) => {
+  const formatedValue = React.useMemo((): string => {
+    try {
+      const bytes = typeof value === 'string' ? parseInt(value, 10) : value;
+      if (Number.isNaN(bytes)) return `-Bytes`;
+      if (!bytes || bytes < 1024) return `${Math.ceil(bytes || 0)}${sizes[0]}`;
+      const pow = Math.floor(Math.log2(bytes) / 10);
+      const multiplier = 10 ** (precision < 0 ? 0 : precision);
+      return (
+        Math.round((bytes * multiplier) / 1024 ** pow) / multiplier + sizes[pow]
+      );
+    } catch (e) {
+      return `-Bytes`;
+    }
   }, [value]);
 
   return <span>{formatedValue}</span>;

kafka-ui-react-app/src/components/common/BytesFormatted/__tests__/BytesFormatted.spec.tsx

@@ -0,0 +1,38 @@
+import { shallow } from 'enzyme';
+import React from 'react';
+import BytesFormatted, { sizes } from '../BytesFormatted';
+
+describe('BytesFormatted', () => {
+  it('renders Bytes correctly', () => {
+    const component = shallow(<BytesFormatted value={666} />);
+    expect(component.text()).toEqual('666Bytes');
+  });
+
+  it('renders correct units', () => {
+    let value = 1;
+    sizes.forEach((unit) => {
+      const component = shallow(<BytesFormatted value={value} />);
+      expect(component.text()).toEqual(`1${unit}`);
+      value *= 1024;
+    });
+  });
+
+  it('renders correct precision', () => {
+    let component = shallow(<BytesFormatted value={2000} precision={100} />);
+    expect(component.text()).toEqual(`1.953125${sizes[1]}`);
+
+    component = shallow(<BytesFormatted value={10000} precision={5} />);
+    expect(component.text()).toEqual(`9.76563${sizes[1]}`);
+  });
+
+  it('correctly handles invalid props', () => {
+    let component = shallow(<BytesFormatted value={10000} precision={-1} />);
+    expect(component.text()).toEqual(`10${sizes[1]}`);
+
+    component = shallow(<BytesFormatted value="some string" />);
+    expect(component.text()).toEqual(`-${sizes[0]}`);
+
+    component = shallow(<BytesFormatted value={undefined} />);
+    expect(component.text()).toEqual(`0${sizes[0]}`);
+  });
+});

kafka-ui-react-app/src/components/common/Dashboard/__tests__/Indicator.spec.tsx

@@ -0,0 +1,15 @@
+import { mount } from 'enzyme';
+import React from 'react';
+import Indicator from '../Indicator';
+
+describe('Indicator', () => {
+  it('matches the snapshot', () => {
+    const child = 'Child';
+    const component = mount(
+      <Indicator title="title" label="label">
+        {child}
+      </Indicator>
+    );
+    expect(component).toMatchSnapshot();
+  });
+});

kafka-ui-react-app/src/components/common/Dashboard/__tests__/MetricsWrapper.spec.tsx

@@ -0,0 +1,24 @@
+import { shallow } from 'enzyme';
+import React from 'react';
+import MetricsWrapper from '../MetricsWrapper';
+
+describe('MetricsWrapper', () => {
+  it('correctly adds classes', () => {
+    const className = 'className';
+    const component = shallow(
+      <MetricsWrapper wrapperClassName={className} multiline />
+    );
+    expect(component.find(`.${className}`).exists()).toBeTruthy();
+    expect(component.find('.level-multiline').exists()).toBeTruthy();
+  });
+
+  it('correctly renders children', () => {
+    let component = shallow(<MetricsWrapper />);
+    expect(component.find('.subtitle').exists()).toBeFalsy();
+
+    const title = 'title';
+    component = shallow(<MetricsWrapper title={title} />);
+    expect(component.find('.subtitle').exists()).toBeTruthy();
+    expect(component.text()).toEqual(title);
+  });
+});

kafka-ui-react-app/src/components/common/Dashboard/__tests__/__snapshots__/Indicator.spec.tsx.snap

@@ -0,0 +1,27 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`Indicator matches the snapshot 1`] = `
+<Indicator
+  label="label"
+  title="title"
+>
+  <div
+    className="level-item level-left"
+  >
+    <div
+      title="title"
+    >
+      <p
+        className="heading"
+      >
+        label
+      </p>
+      <p
+        className="title"
+      >
+        Child
+      </p>
+    </div>
+  </div>
+</Indicator>
+`;

kafka-ui-react-app/src/components/common/PageLoader/__tests__/PageLoader.spec.tsx

@@ -0,0 +1,10 @@
+import { mount } from 'enzyme';
+import React from 'react';
+import PageLoader from '../PageLoader';
+
+describe('PageLoader', () => {
+  it('matches the snapshot', () => {
+    const component = mount(<PageLoader />);
+    expect(component).toMatchSnapshot();
+  });
+});

kafka-ui-react-app/src/components/common/PageLoader/__tests__/__snapshots__/PageLoader.spec.tsx.snap

@@ -0,0 +1,36 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`PageLoader matches the snapshot 1`] = `
+<PageLoader>
+  <section
+    className="hero is-fullheight-with-navbar"
+  >
+    <div
+      className="hero-body has-text-centered"
+      style={
+        Object {
+          "justifyContent": "center",
+        }
+      }
+    >
+      <div
+        style={
+          Object {
+            "width": 300,
+          }
+        }
+      >
+        <div
+          className="subtitle"
+        >
+          Loading...
+        </div>
+        <progress
+          className="progress is-small is-primary is-inline-block"
+          max="100"
+        />
+      </div>
+    </div>
+  </section>
+</PageLoader>
+`;