Hello there joschi! 👋

Thank you and congratulations 🎉 for opening your very first issue in this project! 💖

In case you want to claim this issue, please comment down below! We will try to get back to you as soon as we can. 👀 Hey Jochen, thanks for raising the issue.

We've recently changed this behavior (#2755). If you suggest that it should be implied that internal topics start with *double* underscore, shouldn't we count "_schemas" (single underscore) as internal topic for example?
I'm not sure about correct behavior here. Also there's an "internal" flag in kafka via AC, which we use as well (you can see tha pr for #2755). > If you suggest that it should be implied that internal topics start with _double_ underscore, shouldn't we count "_schemas" (single underscore) as internal topic for example?

A single underscore as an indicator is fine (topics starting with two underscores also start with 1 underscore 😉).

We're using Amazon MSK and MSK Connect which is using the convention with two underscore characters as an indicator for internal topics.

Confluent also seems to follow the convention with a single underscore character:
https://docs.confluent.io/platform/current/control-center/topics/overview.html#topics-page

> Internal topics names start with an underscore (_) and should not be individually modified. Modifying an internal topic could adversely impact your Confluent Platform installation and result in unexpected behavior.

The Javadoc of [``](https://kafka.apache.org/33/javadoc/org/apache/kafka/clients/admin/TopicDescription.html#isInternal--) also hints to a similar convention:

> Whether the topic is internal to Kafka. An example of an internal topic is the offsets and group management topic: `__consumer_offsets`.

And here's another indicator:
https://github.com/apache/kafka/blob/3.3.1/clients/src/main/java/org/apache/kafka/common/internals/Topic.java#L29-L36 This issue has been automatically marked as stale because no requested feedback has been provided. It will be closed if no further activity occurs. Thank you for your contributions. The internality flag is being defaulted to a single underscore character now. Can be overridden with "kafka_clusters_internalTopicPrefix" property.
add @Nullable annotation, or do defaulting on higher level (in service) Done, the first way
"Show Internal Topics" not working
TODO: make the "internal" flag prefix configurable, default it to a single underscore character

///

**Describe the bug**

After updating to Kafka UI 0.5.0 (using the Docker image [provectuslabs/kafka-ui:v0.5.0](https://hub.docker.com/layers/provectuslabs/kafka-ui/v0.5.0/images/sha256-cd0fdd18d7d18dd2ba86c237507d7979778405de3e74377fed7582e6a60297d1?context=explore)), the "Show Internal Topics" toggle in the "Topics" view isn't working anymore.

Internal topics are always shown, no matter the state of the toggle.

**Set up**

Docker image [provectuslabs/kafka-ui:v0.5.0](https://hub.docker.com/layers/provectuslabs/kafka-ui/v0.5.0/images/sha256-cd0fdd18d7d18dd2ba86c237507d7979778405de3e74377fed7582e6a60297d1?context=explore) deployed on Kubernetes (no Helm), using the configuration from Kafka UI 0.4.0.

> check commit hash in the top left corner in UI

There is no commit hash shown in the web UI anymore.

**Steps to Reproduce**
<!-- We'd like you to provide an example setup (via docker-compose, helm, etc.) 
to reproduce the problem, especially with a complex setups. -->
Steps to reproduce the behavior:

1. Open Kafka UI 0.5.0.
2. Click on the configured Kafka cluster.
3. Go to "Topics" page.
4. Click on "Show Internal Topics" toggle.

**Expected behavior**

Internal topics (starting with two underscores: `__`) should be hidden when the "Show Internal Topics" toggle is disabled.

**Screenshots**

<img width="1074" alt="Show Internal Topics disabled" src="https://user-images.githubusercontent.com/43951/209367014-c916520e-0207-4275-acd6-9c88517d2d0e.png">
<img width="1079" alt="Show Internal Topics enabled" src="https://user-images.githubusercontent.com/43951/209367073-d5518e1f-cbef-48fc-8580-325c325c02f4.png">
kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java
kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/TopicsService.java kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java
kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java
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 2cd5e0e69cd..7b6b494ad12 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
@@ -25,6 +25,8 @@ public class ClustersProperties {
 
   List<Cluster> clusters = new ArrayList<>();
 
+  String internalTopicPrefix;
+
   @Data
   public static class Cluster {
     String name;
diff --git a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java
index efabd15f098..43a6012d215 100644
--- a/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java
+++ b/kafka-ui-api/src/main/java/com/provectus/kafka/ui/model/InternalTopic.java
@@ -1,9 +1,11 @@
 package com.provectus.kafka.ui.model;
 
+import com.provectus.kafka.ui.config.ClustersProperties;
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 import lombok.Builder;
 import lombok.Data;
 import org.apache.kafka.clients.admin.ConfigEntry;
@@ -14,6 +16,8 @@
 @Builder(toBuilder = true)
 public class InternalTopic {
 
+  ClustersProperties clustersProperties;
+
   // from TopicDescription
   private final String name;
   private final boolean internal;
@@ -40,9 +44,17 @@ public static InternalTopic from(TopicDescription topicDescription,
                                    List<ConfigEntry> configs,
                                    InternalPartitionsOffsets partitionsOffsets,
                                    Metrics metrics,
-                                   InternalLogDirStats logDirInfo) {
+                                   InternalLogDirStats logDirInfo,
+                                   @Nullable String internalTopicPrefix) {
     var topic = InternalTopic.builder();
-    topic.internal(topicDescription.isInternal());
+
+    internalTopicPrefix = internalTopicPrefix == null || internalTopicPrefix.isEmpty()
+        ? "_"
+        : internalTopicPrefix;
+
+    topic.internal(
+        topicDescription.isInternal() || topicDescription.name().startsWith(internalTopicPrefix)
+    );
     topic.name(topicDescription.name());
 
     List<InternalPartition> partitions = topicDescription.partitions().stream()
@@ -56,10 +68,10 @@ public static InternalTopic from(TopicDescription topicDescription,
           List<InternalReplica> replicas = partition.replicas().stream()
               .map(r ->
                   InternalReplica.builder()
-                    .broker(r.id())
-                    .inSync(partition.isr().contains(r))
-                    .leader(partition.leader() != null && partition.leader().id() == r.id())
-                    .build())
+                      .broker(r.id())
+                      .inSync(partition.isr().contains(r))
+                      .leader(partition.leader() != null && partition.leader().id() == r.id())
+                      .build())
               .collect(Collectors.toList());
           partitionDto.replicas(replicas);
 
@@ -79,7 +91,7 @@ public static InternalTopic from(TopicDescription topicDescription,
 
           return partitionDto.build();
         })
-        .collect(Collectors.toList());
+        .toList();
 
     topic.partitions(partitions.stream().collect(
         Collectors.toMap(InternalPartition::getPartition, t -> t)));
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 b172e2b4c61..9aaff3e9ef0 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
@@ -3,6 +3,7 @@
 import static java.util.stream.Collectors.toList;
 import static java.util.stream.Collectors.toMap;
 
+import com.provectus.kafka.ui.config.ClustersProperties;
 import com.provectus.kafka.ui.exception.TopicMetadataException;
 import com.provectus.kafka.ui.exception.TopicNotFoundException;
 import com.provectus.kafka.ui.exception.TopicRecreationException;
@@ -52,6 +53,7 @@ public class TopicsService {
 
   private final AdminClientService adminClientService;
   private final StatisticsCache statisticsCache;
+  private final ClustersProperties clustersProperties;
   @Value("${topic.recreate.maxRetries:15}")
   private int recreateMaxRetries;
   @Value("${topic.recreate.delay.seconds:1}")
@@ -127,7 +129,8 @@ private List<InternalTopic> createList(List<String> orderedNames,
             configs.getOrDefault(t, List.of()),
             partitionsOffsets,
             metrics,
-            logDirInfo
+            logDirInfo,
+            clustersProperties.getInternalTopicPrefix()
         ))
         .collect(toList());
   }
@@ -459,7 +462,9 @@ public Mono<List<InternalTopic>> getTopicsForPagination(KafkaCluster cluster) {
                     stats.getTopicConfigs().getOrDefault(topicName, List.of()),
                     InternalPartitionsOffsets.empty(),
                     stats.getMetrics(),
-                    stats.getLogDirInfo()))
+                    stats.getLogDirInfo(),
+                    clustersProperties.getInternalTopicPrefix()
+                    ))
             .collect(toList())
         );
   }
diff --git a/kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java b/kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java
index 499d5fe965a..35f4c9b1b41 100644
--- a/kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java
+++ b/kafka-ui-e2e-checks/src/main/java/com/provectus/kafka/ui/pages/topics/TopicsList.java
@@ -6,7 +6,6 @@
 import com.provectus.kafka.ui.pages.BasePage;
 import io.qameta.allure.Step;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -175,6 +174,12 @@ public TopicGridItem getTopicItem(String name) {
                 .findFirst().orElseThrow();
     }
 
+    @Step
+    public TopicGridItem getAnyNonInternalTopic() {
+        return getNonInternalTopics().stream()
+                .findAny().orElseThrow();
+    }
+
     @Step
     public List<TopicGridItem> getNonInternalTopics() {
         return initGridItems().stream()
@@ -207,8 +212,7 @@ public TopicsList selectItem(boolean select) {
         public boolean isInternal() {
             boolean internal = false;
             try {
-                element.$x("./td[2]/a/span").shouldBe(visible, Duration.ofMillis(500));
-                internal = true;
+                internal = element.$x("./td[2]/a/span").isDisplayed();
             } catch (Throwable ignored) {
             }
             return internal;

diff --git a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java
index 44b771f1542..8867400fee3 100644
--- a/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java
+++ b/kafka-ui-api/src/test/java/com/provectus/kafka/ui/service/TopicsServicePaginationTest.java
@@ -69,7 +69,7 @@ public void shouldListFirst25Topics() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -95,7 +95,7 @@ public void shouldListFirst25TopicsSortedByNameDescendingOrder() {
         .map(Objects::toString)
         .map(name -> new TopicDescription(name, false, List.of()))
         .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-            Metrics.empty(), InternalLogDirStats.empty()))
+            Metrics.empty(), InternalLogDirStats.empty(), "_"))
         .collect(Collectors.toMap(InternalTopic::getName, Function.identity()));
     init(internalTopics);
 
@@ -122,7 +122,7 @@ public void shouldCalculateCorrectPageCountForNonDivisiblePageSize() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -141,7 +141,7 @@ public void shouldCorrectlyHandleNonPositivePageNumberAndPageSize() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -160,7 +160,7 @@ public void shouldListBotInternalAndNonInternalTopics() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, Integer.parseInt(name) % 10 == 0, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -181,7 +181,7 @@ public void shouldListOnlyNonInternalTopics() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, Integer.parseInt(name) % 5 == 0, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -202,7 +202,7 @@ public void shouldListOnlyTopicsContainingOne() {
             .map(Objects::toString)
             .map(name -> new TopicDescription(name, false, List.of()))
             .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), null,
-                Metrics.empty(), InternalLogDirStats.empty()))
+                Metrics.empty(), InternalLogDirStats.empty(), "_"))
             .collect(Collectors.toMap(InternalTopic::getName, Function.identity()))
     );
 
@@ -224,7 +224,7 @@ public void shouldListTopicsOrderedByPartitionsCount() {
                     new TopicPartitionInfo(p, null, List.of(), List.of()))
                 .collect(Collectors.toList())))
         .map(topicDescription -> InternalTopic.from(topicDescription, List.of(), InternalPartitionsOffsets.empty(),
-            Metrics.empty(), InternalLogDirStats.empty()))
+            Metrics.empty(), InternalLogDirStats.empty(), "_"))
         .collect(Collectors.toMap(InternalTopic::getName, Function.identity()));
 
     init(internalTopics);
diff --git a/kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java b/kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java
index 3f1347cb391..92e580ec32c 100644
--- a/kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java
+++ b/kafka-ui-e2e-checks/src/test/java/com/provectus/kafka/ui/smokeSuite/topics/TopicsTest.java
@@ -52,7 +52,8 @@ public class TopicsTest extends BaseTest {
             .setMaxSizeOnDisk(NOT_SET);
     private static final Topic TOPIC_FOR_CHECK_FILTERS = new Topic()
             .setName("topic-for-check-filters-" + randomAlphabetic(5));
-    private static final Topic TOPIC_FOR_DELETE = new Topic().setName("topic-to-delete-" + randomAlphabetic(5));
+    private static final Topic TOPIC_FOR_DELETE = new Topic()
+            .setName("topic-to-delete-" + randomAlphabetic(5));
     private static final List<Topic> TOPIC_LIST = new ArrayList<>();
 
     @BeforeClass(alwaysRun = true)
@@ -89,11 +90,11 @@ public void createTopic() {
     void checkAvailableOperations() {
         navigateToTopics();
         topicsList
-                .getTopicItem("my_ksql_1ksql_processing_log")
+                .getTopicItem(TOPIC_TO_UPDATE_AND_DELETE.getName())
                 .selectItem(true);
         verifyElementsCondition(topicsList.getActionButtons(), Condition.enabled);
         topicsList
-                .getTopicItem("_confluent-ksql-my_ksql_1_command_topic")
+                .getTopicItem(TOPIC_FOR_CHECK_FILTERS.getName())
                 .selectItem(true);
         Assert.assertFalse(topicsList.isCopySelectedTopicBtnEnabled(), "isCopySelectedTopicBtnEnabled()");
     }
@@ -456,7 +457,7 @@ public void checkCopyTopicPossibility() {
                 .setNumberOfPartitions(1);
         navigateToTopics();
         topicsList
-                .getTopicItem("_schemas")
+                .getAnyNonInternalTopic()
                 .selectItem(true)
                 .clickCopySelectedTopicBtn();
         topicCreateEditForm
