Merge branch 'master' into vlad/develop

README.md

@@ -199,7 +199,6 @@ For example, if you want to use an environment variable to set the `name` parame
 |`KAFKA_CLUSTERS_0_METRICS_PORT`        	 |Open metrics port of a broker
 |`KAFKA_CLUSTERS_0_METRICS_TYPE`        	 |Type of metrics retriever to use. Valid values are JMX (default) or PROMETHEUS. If Prometheus, then metrics are read from prometheus-jmx-exporter instead of jmx
 |`KAFKA_CLUSTERS_0_READONLY`        	|Enable read-only mode. Default: false
-|`KAFKA_CLUSTERS_0_DISABLELOGDIRSCOLLECTION`        	|Disable collecting segments information. It should be true for confluent cloud. Default: false
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME` |Given name for the Kafka Connect cluster
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS` |Address of the Kafka Connect service endpoint
 |`KAFKA_CLUSTERS_0_KAFKACONNECT_0_USERNAME`| Kafka Connect cluster's basic authentication username

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

@@ -38,7 +38,6 @@ public class ClustersProperties {
     MetricsConfigData metrics;
     Properties properties;
     boolean readOnly = false;
-    boolean disableLogDirsCollection = false;
     List<SerdeConfig> serde = new ArrayList<>();
     String defaultKeySerde;
     String defaultValueSerde;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/TopicsController.java

@@ -175,7 +175,7 @@ public class TopicsController extends AbstractController implements TopicsApi {
           List<InternalTopic> filtered = existingTopics.stream()
               .filter(topic -> !topic.isInternal()
                   || showInternal != null && showInternal)
-              .filter(topic -> search == null || StringUtils.contains(topic.getName(), search))
+              .filter(topic -> search == null || StringUtils.containsIgnoreCase(topic.getName(), search))
               .sorted(comparator)
               .toList();
           var totalPages = (filtered.size() / pageSize)

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

@@ -26,7 +26,6 @@ public class KafkaCluster {
   private final String bootstrapServers;
   private final Properties properties;
   private final boolean readOnly;
-  private final boolean disableLogDirsCollection;
   private final MetricsConfig metricsConfig;
   private final DataMasking masking;
   private final Supplier<PollingThrottler> throttler;

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

@@ -39,7 +39,6 @@ public class KafkaClusterFactory {
     builder.bootstrapServers(clusterProperties.getBootstrapServers());
     builder.properties(Optional.ofNullable(clusterProperties.getProperties()).orElse(new Properties()));
     builder.readOnly(clusterProperties.isReadOnly());
-    builder.disableLogDirsCollection(clusterProperties.isDisableLogDirsCollection());
     builder.masking(DataMasking.create(clusterProperties.getMasking()));
     builder.metricsConfig(metricsConfigDataToMetricsConfig(clusterProperties.getMetrics()));
     builder.throttler(PollingThrottler.throttlerSupplier(clusterProperties));

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

@@ -66,6 +66,7 @@ import org.apache.kafka.common.errors.GroupIdNotFoundException;
 import org.apache.kafka.common.errors.GroupNotEmptyException;
 import org.apache.kafka.common.errors.InvalidRequestException;
 import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
 import org.apache.kafka.common.requests.DescribeLogDirsResponse;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -297,7 +298,12 @@ public class ReactiveAdminClient implements Closeable {
 
   public Mono<Map<Integer, Map<String, DescribeLogDirsResponse.LogDirInfo>>> describeLogDirs(
       Collection<Integer> brokerIds) {
-    return toMono(client.describeLogDirs(brokerIds).all());
+    return toMono(client.describeLogDirs(brokerIds).all())
+        .onErrorResume(UnsupportedVersionException.class, th -> Mono.just(Map.of()))
+        .onErrorResume(th -> true, th -> {
+          log.warn("Error while calling describeLogDirs", th);
+          return Mono.just(Map.of());
+        });
   }
 
   public Mono<ClusterDescription> describeCluster() {

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

@@ -1,5 +1,7 @@
 package com.provectus.kafka.ui.service;
 
+import static com.provectus.kafka.ui.service.ReactiveAdminClient.ClusterDescription;
+
 import com.provectus.kafka.ui.model.Feature;
 import com.provectus.kafka.ui.model.InternalLogDirStats;
 import com.provectus.kafka.ui.model.KafkaCluster;
@@ -9,10 +11,12 @@ import com.provectus.kafka.ui.model.Statistics;
 import com.provectus.kafka.ui.service.metrics.MetricsCollector;
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.admin.ConfigEntry;
 import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.Node;
 import org.springframework.stereotype.Service;
 import reactor.core.publisher.Mono;
 
@@ -21,7 +25,7 @@ import reactor.core.publisher.Mono;
 @Slf4j
 public class StatisticsService {
 
-  private final MetricsCollector metricsClusterUtil;
+  private final MetricsCollector metricsCollector;
   private final AdminClientService adminClientService;
   private final FeatureService featureService;
   private final StatisticsCache cache;
@@ -35,8 +39,8 @@ public class StatisticsService {
             ac.describeCluster().flatMap(description ->
                 Mono.zip(
                     List.of(
-                        metricsClusterUtil.getBrokerMetrics(cluster, description.getNodes()),
-                        getLogDirInfo(cluster, ac),
+                        metricsCollector.getBrokerMetrics(cluster, description.getNodes()),
+                        getLogDirInfo(description, ac),
                         featureService.getAvailableFeatures(cluster, description.getController()),
                         loadTopicConfigs(cluster),
                         describeTopics(cluster)),
@@ -58,11 +62,9 @@ public class StatisticsService {
             e -> Mono.just(Statistics.empty().toBuilder().lastKafkaException(e).build()));
   }
 
-  private Mono<InternalLogDirStats> getLogDirInfo(KafkaCluster cluster, ReactiveAdminClient c) {
-    if (!cluster.isDisableLogDirsCollection()) {
-      return c.describeLogDirs().map(InternalLogDirStats::new);
-    }
-    return Mono.just(InternalLogDirStats.empty());
+  private Mono<InternalLogDirStats> getLogDirInfo(ClusterDescription desc, ReactiveAdminClient ac) {
+    var brokerIds = desc.getNodes().stream().map(Node::id).collect(Collectors.toSet());
+    return ac.describeLogDirs(brokerIds).map(InternalLogDirStats::new);
   }
 
   private Mono<Map<String, TopicDescription>> describeTopics(KafkaCluster c) {

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

@@ -61,12 +61,12 @@ class MessagesServiceTest extends AbstractIntegrationTest {
   }
 
   @Test
-  void maskingAppliedOnConfiguredClusters() {
+  void maskingAppliedOnConfiguredClusters() throws Exception {
     String testTopic = MASKED_TOPICS_PREFIX + UUID.randomUUID();
     try (var producer = KafkaTestProducer.forKafka(kafka)) {
       createTopic(new NewTopic(testTopic, 1, (short) 1));
       producer.send(testTopic, "message1");
-      producer.send(testTopic, "message2");
+      producer.send(testTopic, "message2").get();
 
       Flux<TopicMessageDTO> msgsFlux = messagesService.loadMessages(
           cluster,
@@ -91,4 +91,4 @@ class MessagesServiceTest extends AbstractIntegrationTest {
     }
   }
 
-}
+}

kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/models/Topic.java

@@ -3,6 +3,7 @@ package com.provectus.kafka.ui.models;
 import com.provectus.kafka.ui.pages.topic.enums.CleanupPolicyValue;
 import com.provectus.kafka.ui.pages.topic.enums.CustomParameterType;
 import com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk;
+import com.provectus.kafka.ui.pages.topic.enums.TimeToRetain;
 import lombok.Data;
 import lombok.experimental.Accessors;
 
@@ -14,4 +15,5 @@ public class Topic {
     private CustomParameterType customParameterType;
     private CleanupPolicyValue cleanupPolicyValue;
     private MaxSizeOnDisk maxSizeOnDisk;
+    private TimeToRetain timeToRetain;
 }

kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topic/TopicCreateEditForm.java

@@ -13,6 +13,7 @@ import com.provectus.kafka.ui.pages.BasePage;
 import com.provectus.kafka.ui.pages.topic.enums.CleanupPolicyValue;
 import com.provectus.kafka.ui.pages.topic.enums.CustomParameterType;
 import com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk;
+import com.provectus.kafka.ui.pages.topic.enums.TimeToRetain;
 import io.qameta.allure.Step;
 
 public class TopicCreateEditForm extends BasePage {
@@ -30,6 +31,8 @@ public class TopicCreateEditForm extends BasePage {
   protected SelenideElement customParameterValueField = $x("//input[@placeholder='Value']");
   protected SelenideElement validationCustomParameterValueMsg = $x("//p[contains(text(),'Value is required')]");
   protected String ddlElementLocator = "//li[@value='%s']";
+  protected String btnTimeToRetainLocator = "//button[@class][text()='%s']";
+
 
   @Step
   public TopicCreateEditForm waitUntilScreenReady() {
@@ -46,6 +49,10 @@ public class TopicCreateEditForm extends BasePage {
     return isEnabled(deleteCustomParameterBtn);
   }
 
+  public boolean isNameFieldEnabled(){
+    return isEnabled(nameField);
+  }
+
   @Step
   public TopicCreateEditForm setTopicName(String topicName) {
     nameField.shouldBe(Condition.enabled).clear();
@@ -118,13 +125,8 @@ public class TopicCreateEditForm extends BasePage {
   }
 
   @Step
-  public TopicCreateEditForm setTimeToRetainDataInMsUsingButtons(String value) {
-    timeToRetainField
-        .parent()
-        .parent()
-        .$$("button")
-        .find(Condition.exactText(value))
-        .click();
+  public TopicCreateEditForm setTimeToRetainDataByButtons(TimeToRetain timeToRetain) {
+    $x(String.format(btnTimeToRetainLocator, timeToRetain.getButton())).shouldBe(Condition.enabled).click();
     return this;
   }
 

kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topic/enums/TimeToRetain.java

@@ -0,0 +1,25 @@
+package com.provectus.kafka.ui.pages.topic.enums;
+
+public enum TimeToRetain {
+  BTN_12_HOURS("12 hours", "43200000"),
+  BTN_1_DAY("1 day", "86400000"),
+  BTN_2_DAYS("2 days", "172800000"),
+  BTN_7_DAYS("7 days", "604800000"),
+  BTN_4_WEEKS("4 weeks", "2419200000");
+
+  private final String button;
+  private final String value;
+
+  TimeToRetain(String button, String value) {
+    this.button = button;
+    this.value = value;
+  }
+
+  public String getButton(){
+    return button;
+  }
+
+  public String getValue(){
+    return value;
+  }
+}

kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/suite/topics/TopicsTests.java

@@ -8,7 +8,9 @@ import static com.provectus.kafka.ui.pages.topic.enums.CleanupPolicyValue.DELETE
 import static com.provectus.kafka.ui.pages.topic.enums.CustomParameterType.COMPRESSION_TYPE;
 import static com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk.NOT_SET;
 import static com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk.SIZE_1_GB;
-import static com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk.SIZE_20_GB;
+import static com.provectus.kafka.ui.pages.topic.enums.MaxSizeOnDisk.SIZE_50_GB;
+import static com.provectus.kafka.ui.pages.topic.enums.TimeToRetain.BTN_2_DAYS;
+import static com.provectus.kafka.ui.pages.topic.enums.TimeToRetain.BTN_7_DAYS;
 import static org.apache.commons.lang.RandomStringUtils.randomAlphabetic;
 import static org.apache.commons.lang3.RandomUtils.nextInt;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -50,10 +52,10 @@ public class TopicsTests extends BaseTest {
   private static final Topic TOPIC_TO_UPDATE_AND_DELETE = new Topic()
       .setName("topic-to-update-and-delete-" + randomAlphabetic(5))
       .setNumberOfPartitions(1)
-      .setCleanupPolicyValue(COMPACT)
-      .setTimeToRetainData("604800001")
-      .setMaxSizeOnDisk(SIZE_20_GB)
-      .setMaxMessageBytes("1000020")
+      .setCleanupPolicyValue(DELETE)
+      .setTimeToRetain(BTN_7_DAYS)
+      .setMaxSizeOnDisk(NOT_SET)
+      .setMaxMessageBytes("1048588")
       .setMessageKey(randomAlphabetic(5))
       .setMessageContent(randomAlphabetic(10));
   private static final Topic TOPIC_TO_CHECK_SETTINGS = new Topic()
@@ -132,24 +134,43 @@ public class TopicsTests extends BaseTest {
         .openDotMenu()
         .clickEditSettingsMenu();
     topicCreateEditForm
-        .waitUntilScreenReady()
+        .waitUntilScreenReady();
+    SoftAssertions softly = new SoftAssertions();
+    softly.assertThat(topicCreateEditForm.getCleanupPolicy()).as("getCleanupPolicy()")
+        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getCleanupPolicyValue().getVisibleText());
+    softly.assertThat(topicCreateEditForm.getTimeToRetain()).as("getTimeToRetain()")
+        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getTimeToRetain().getValue());
+    softly.assertThat(topicCreateEditForm.getMaxSizeOnDisk()).as("getMaxSizeOnDisk()")
+        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getMaxSizeOnDisk().getVisibleText());
+    softly.assertThat(topicCreateEditForm.getMaxMessageBytes()).as("getMaxMessageBytes()")
+        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getMaxMessageBytes());
+    softly.assertAll();
+    TOPIC_TO_UPDATE_AND_DELETE
+        .setCleanupPolicyValue(COMPACT)
+        .setTimeToRetain(BTN_2_DAYS)
+        .setMaxSizeOnDisk(SIZE_50_GB).setMaxMessageBytes("1048589");
+    topicCreateEditForm
         .selectCleanupPolicy((TOPIC_TO_UPDATE_AND_DELETE.getCleanupPolicyValue()))
-        .setMinInsyncReplicas(10)
-        .setTimeToRetainDataInMs(TOPIC_TO_UPDATE_AND_DELETE.getTimeToRetainData())
+        .setTimeToRetainDataByButtons(TOPIC_TO_UPDATE_AND_DELETE.getTimeToRetain())
         .setMaxSizeOnDiskInGB(TOPIC_TO_UPDATE_AND_DELETE.getMaxSizeOnDisk())
         .setMaxMessageBytes(TOPIC_TO_UPDATE_AND_DELETE.getMaxMessageBytes())
         .clickCreateTopicBtn();
+    softly.assertThat(topicDetails.isAlertWithMessageVisible(SUCCESS, "Topic successfully updated."))
+        .as("isAlertWithMessageVisible()").isTrue();
+    softly.assertThat(topicDetails.isTopicHeaderVisible(TOPIC_TO_UPDATE_AND_DELETE.getName()))
+        .as("isTopicHeaderVisible()").isTrue();
+    softly.assertAll();
     topicDetails
         .waitUntilScreenReady();
     navigateToTopicsAndOpenDetails(TOPIC_TO_UPDATE_AND_DELETE.getName());
     topicDetails
         .openDotMenu()
         .clickEditSettingsMenu();
-    SoftAssertions softly = new SoftAssertions();
+    softly.assertThat(topicCreateEditForm.isNameFieldEnabled()).as("isNameFieldEnabled()").isFalse();
     softly.assertThat(topicCreateEditForm.getCleanupPolicy()).as("getCleanupPolicy()")
         .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getCleanupPolicyValue().getVisibleText());
     softly.assertThat(topicCreateEditForm.getTimeToRetain()).as("getTimeToRetain()")
-        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getTimeToRetainData());
+        .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getTimeToRetain().getValue());
     softly.assertThat(topicCreateEditForm.getMaxSizeOnDisk()).as("getMaxSizeOnDisk()")
         .isEqualTo(TOPIC_TO_UPDATE_AND_DELETE.getMaxSizeOnDisk().getVisibleText());
     softly.assertThat(topicCreateEditForm.getMaxMessageBytes()).as("getMaxMessageBytes()")

kafka-ui-react-app/package.json

@@ -26,6 +26,7 @@
     "jest": "^29.0.3",
     "jest-watch-typeahead": "^2.0.0",
     "json-schema-faker": "^0.5.0-rcv.44",
+    "jsonpath-plus": "^7.2.0",
     "lodash": "^4.17.21",
     "pretty-ms": "7.0.1",
     "react": "^18.1.0",

kafka-ui-react-app/pnpm-lock.yaml

@@ -63,6 +63,7 @@ specifiers:
   jest-styled-components: ^7.0.8
   jest-watch-typeahead: ^2.0.0
   json-schema-faker: ^0.5.0-rcv.44
+  jsonpath-plus: ^7.2.0
   lint-staged: ^13.0.2
   lodash: ^4.17.21
   prettier: ^2.3.1
@@ -116,6 +117,7 @@ dependencies:
   jest: 29.0.3_yqiaopbgmqcuvx27p5xxvum6wm
   jest-watch-typeahead: 2.0.0_jest@29.0.3
   json-schema-faker: 0.5.0-rcv.44
+  jsonpath-plus: 7.2.0
   lodash: 4.17.21
   pretty-ms: 7.0.1
   react: 18.1.0
@@ -7058,6 +7060,11 @@ packages:
     engines: {node: '>=10.0.0'}
     dev: false
 
+  /jsonpath-plus/7.2.0:
+    resolution: {integrity: sha512-zBfiUPM5nD0YZSBT/o/fbCUlCcepMIdP0CJZxM1+KgA4f2T206f6VAg9e7mX35+KlMaIc5qXW34f3BnwJ3w+RA==}
+    engines: {node: '>=12.0.0'}
+    dev: false
+
   /jsx-ast-utils/3.3.0:
     resolution: {integrity: sha512-XzO9luP6L0xkxwhIJMTJQpZo/eeN60K08jHdexfD569AGxeNug6UketeHXEhROoM8aR7EcUoOQmIhcJQjcuq8Q==}
     engines: {node: '>=4.0'}

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

@@ -1,4 +1,3 @@
-import get from 'lodash/get';
 import React from 'react';
 import styled from 'styled-components';
 import useDataSaver from 'lib/hooks/useDataSaver';
@@ -7,6 +6,7 @@ import MessageToggleIcon from 'components/common/Icons/MessageToggleIcon';
 import IconButtonWrapper from 'components/common/Icons/IconButtonWrapper';
 import { Dropdown, DropdownItem } from 'components/common/Dropdown';
 import { formatTimestamp } from 'lib/dateTimeHelpers';
+import { JSONPath } from 'jsonpath-plus';
 
 import MessageContent from './MessageContent/MessageContent';
 import * as S from './MessageContent/MessageContent.styled';
@@ -88,9 +88,12 @@ const Message: React.FC<Props> = ({
     return (
       <>
         {filters.map((item) => (
-          <div>
-            {item.field}: {get(parsedJson, item.path)}
-          </div>
+          <span key={`${item.path}--${item.field}`}>
+            {item.field}:{' '}
+            {JSON.stringify(
+              JSONPath({ path: item.path, json: parsedJson, wrap: false })
+            )}
+          </span>
         ))}
       </>
     );
@@ -116,7 +119,7 @@ const Message: React.FC<Props> = ({
         <StyledDataCell title={key}>
           {renderFilteredJson(key, keyFilters)}
         </StyledDataCell>
-        <StyledDataCell>
+        <StyledDataCell title={content}>
           <S.Metadata>
             <S.MetadataValue>
               {renderFilteredJson(content, contentFilters)}

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

@@ -52,7 +52,18 @@ const MessagesTable: React.FC = () => {
   };
 
   return (
-    <>
+    <div style={{ position: 'relative' }}>
+      {previewFor !== null && (
+        <PreviewModal
+          values={previewFor === 'key' ? keyFilters : contentFilters}
+          toggleIsOpen={() => setPreviewFor(null)}
+          setFilters={(payload: PreviewFilter[]) =>
+            previewFor === 'key'
+              ? setKeyFilters(payload)
+              : setContentFilters(payload)
+          }
+        />
+      )}
       <Table isFullwidth>
         <thead>
           <tr>
@@ -77,18 +88,6 @@ const MessagesTable: React.FC = () => {
               onPreview={() => setPreviewFor('content')}
             />
             <TableHeaderCell> </TableHeaderCell>
-
-            {previewFor !== null && (
-              <PreviewModal
-                values={previewFor === 'key' ? keyFilters : contentFilters}
-                toggleIsOpen={() => setPreviewFor(null)}
-                setFilters={(payload: PreviewFilter[]) =>
-                  previewFor === 'key'
-                    ? setKeyFilters(payload)
-                    : setContentFilters(payload)
-                }
-              />
-            )}
           </tr>
         </thead>
         <tbody>
@@ -139,7 +138,7 @@ const MessagesTable: React.FC = () => {
           </Button>
         </S.Pages>
       </S.Pagination>
-    </>
+    </div>
   );
 };
 

kafka-ui-react-app/src/components/Topics/Topic/Messages/PreviewModal.styled.ts

@@ -7,6 +7,7 @@ export const PreviewModal = styled.div`
   background: ${({ theme }) => theme.modal.backgroundColor};
   position: absolute;
   left: 25%;
+  top: 30px; // some margin
   border: 1px solid ${({ theme }) => theme.modal.border.contrast};
   box-shadow: ${({ theme }) => theme.modal.shadow};
   padding: 32px;

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

@@ -2,9 +2,15 @@ import styled from 'styled-components';
 
 export const Replica = styled.span.attrs({ 'aria-label': 'replica-info' })<{
   leader?: boolean;
+  outOfSync?: boolean;
 }>`
-  color: ${({ leader, theme }) =>
-    leader ? theme.topicMetaData.liderReplica.color : null};
+  color: ${({ leader, outOfSync, theme }) => {
+    if (outOfSync) return theme.topicMetaData.outOfSync.color;
+    if (leader) return theme.topicMetaData.liderReplica.color;
+    return null;
+  }};
+
+  font-weight: ${({ outOfSync }) => (outOfSync ? '500' : null)};
 
   &:after {
     content: ', ';

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

@@ -51,9 +51,10 @@ const Overview: React.FC = () => {
           if (replicas === undefined || replicas.length === 0) {
             return 0;
           }
-          return replicas?.map(({ broker, leader }: Replica) => (
+          return replicas?.map(({ broker, leader, inSync }: Replica) => (
             <S.Replica
               leader={leader}
+              outOfSync={!inSync}
               key={broker}
               title={leader ? 'Leader' : ''}
             >

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

@@ -75,6 +75,19 @@ describe('Overview', () => {
     );
   });
 
+  describe('when replicas out of sync', () => {
+    it('should be the appropriate color', () => {
+      render(<Replica outOfSync />);
+      const element = screen.getByLabelText('replica-info');
+      expect(element).toBeInTheDocument();
+      expect(element).toHaveStyleRule(
+        'color',
+        theme.topicMetaData.outOfSync.color
+      );
+      expect(element).toHaveStyleRule('font-weight', '500');
+    });
+  });
+
   describe('when it has internal flag', () => {
     it('renders the Action button for Topic', () => {
       renderComponent({

kafka-ui-react-app/src/theme/theme.ts

@@ -517,6 +517,9 @@ const theme = {
     liderReplica: {
       color: Colors.green[60],
     },
+    outOfSync: {
+      color: Colors.red[50],
+    },
   },
   dangerZone: {
     borderColor: Colors.neutral[10],