feat: create system topic (#4229)

* feat: create system topic

* test: creating a system topic with cli

crates/fluvio-cli/src/client/topic/create.rs

@@ -220,6 +220,8 @@ impl CreateTopicOpt {
             topic_spec.set_compression_type(compression_type);
         }
 
+        topic_spec.set_system(self.setting.system);
+
         if self.setting.segment_size.is_some() || self.setting.max_partition_size.is_some() {
             let mut storage = TopicStorageConfig::default();
 
@@ -268,6 +270,11 @@ pub struct TopicConfigOpt {
     /// Ex: `2048`, '2 Ki', '10 MiB', `1 GB`
     #[arg(long, value_name = "bytes")]
     max_partition_size: Option<bytesize::ByteSize>,
+
+    /// Flag to create a system topic
+    /// System topics are for internal operations
+    #[arg(long, short = 's', hide = true)]
+    system: bool,
 }
 
 /// module to load partitions maps from file

tests/cli/fluvio_smoke_tests/topic-basic.bats

@@ -20,6 +20,9 @@ setup_file() {
     TOPIC_NAME_REPLICA=$(random_string)
     export TOPIC_NAME_REPLICA
 
+    TOPIC_NAME_SYSTEM=$(random_string)
+    export TOPIC_NAME_SYSTEM
+
     DEDUP_FILTER_NAME="dedup-filter"
     export DEDUP_FILTER_NAME
 
@@ -219,3 +222,22 @@ EOF
     assert_output --partial "Deduplication Count Bound:10"
     assert_output --partial "Deduplication Age Bound:1"
 }
+
+# Create a system topic
+@test "Create a system topic" {
+    if [ "$FLUVIO_CLI_RELEASE_CHANNEL" == "stable" ]; then
+        skip "don't run on fluvio cli stable version"
+    fi
+    if [ "$FLUVIO_CLUSTER_RELEASE_CHANNEL" == "stable" ]; then
+        skip "don't run on cluster stable version"
+    fi
+    debug_msg "Topic name: $TOPIC_NAME_SYSTEM"
+    run timeout 15s "$FLUVIO_BIN" topic create "$TOPIC_NAME_SYSTEM" --system
+    assert_output --partial "topic "\"$TOPIC_NAME_SYSTEM"\" created"
+    assert_success
+
+    # Check if the topic is a system topic
+    run bash -c 'timeout 15s "$FLUVIO_BIN" partition list --system | grep "$TOPIC_NAME_SYSTEM"'
+    assert_line --partial --index 0 "$TOPIC_NAME_SYSTEM"
+    assert_success
+}