feat: add kafka-replica-verification command (closes #2)

README.md

@@ -79,6 +79,7 @@ Currently, the following commands are supported:
 * `kafka-avro-console-producer`
 * `kafka-verifiable-consumer`
 * `kafka-verifiable-producer`
+* `kafka-replica-verification`
 * `kafka-broker-api-versions`
 * `kafka-consumer-groups`
 * `kafka-delete-records`

kafkashell/constants.py

@@ -24,6 +24,7 @@
 COMMAND_KAFKA_VERIFIABLE_CONSUMER = "kafka-verifiable-consumer"
 COMMAND_KAFKA_VERIFIABLE_PRODUCER = "kafka-verifiable-producer"
 COMMAND_KAFKA_CONSUMER_GROUPS = "kafka-consumer-groups"
+COMMAND_KAFKA_REPLICA_VERIFICATION = "kafka-replica-verification"
 COMMAND_KAFKA_BROKER_API_VERSIONS = "kafka-broker-api-versions"
 COMMAND_KAFKA_DELETE_RECORDS = "kafka-delete-records"
 COMMAND_KAFKA_LOG_DIRS = "kafka-log-dirs"

kafkashell/data/completer.json

@@ -792,6 +792,36 @@
                 }
             }
         },
+        "kafka-replica-verification": {
+            "name": "kafka-replica-verification",
+            "description": "Validate that all replicas for a set of topics have the same data.",
+            "options": {
+                "--broker-list": {
+                    "name": "--broker-list",
+                    "description": "The Kafka broker(s) to connect to. NOTE: This'll overwrite the selected cluster value."
+                },
+                "--fetch-size": {
+                    "name": "--fetch-size",
+                    "description": "The fetch size of each request in bytes. Default: 1048576."
+                },
+                "--max-wait-ms": {
+                    "name": "--max-wait-ms",
+                    "description": "The max amount of time each fetch request waits. Default: 1000."
+                },
+                "--report-interval-ms": {
+                    "name": "--report-interval-ms",
+                    "description": "The reporting interval. Default: 30000."
+                },
+                "--time": {
+                    "name": "--time",
+                    "description": "Timestamp for getting the initial offsets. Default: -1."
+                },
+                "--topic-white-list": {
+                    "name": "--topic-white-list",
+                    "description": "Whitelist of topics to verify replica consistency using Java regex. Default: '.*'."
+                }
+            }
+        },
         "kafka-broker-api-versions": {
             "name": "kafka-broker-api-versions",
             "description": "Retrieve broker version information.",

kafkashell/executor.py

@@ -87,6 +87,9 @@ def execute_valid_command(self, command):
         elif command.startswith(constants.COMMAND_KAFKA_CONSUMER_GROUPS):
             final_command = self.handle_kafka_consumer_groups_command(command)
 
+        elif command.startswith(constants.COMMAND_KAFKA_REPLICA_VERIFICATION):
+            final_command = self.handle_kafka_replica_verification_command(command)
+
         elif command.startswith(constants.COMMAND_KAFKA_BROKER_API_VERSIONS):
             final_command = self.handle_kafka_broker_api_versions_command(command)
 
@@ -190,6 +193,10 @@ def handle_kafka_consumer_groups_command(self, command):
         command += self.handle_admin_client_settings(command)
         return command
 
+    def handle_kafka_replica_verification_command(self, command):
+        command += self.handle_broker_list_flag(command)
+        return command
+
     def handle_kafka_broker_api_versions_command(self, command):
         command += self.handle_bootstrap_server_flag(command)
         command += self.handle_admin_client_settings(command)

tests/test_completer_data.py

@@ -4,7 +4,7 @@
     (
         "",
         ["version", "cluster-select", "cluster-describe", "exit", "clear", "kafka-acls", "kafka-avro-console-consumer",
-         "kafka-avro-console-producer",
+         "kafka-avro-console-producer", "kafka-replica-verification",
          "kafka-broker-api-versions", "kafka-configs", "kafka-console-consumer", "kafka-console-producer",
          "kafka-consumer-groups", "kafka-delete-records", "kafka-dump-log", "kafka-log-dirs", "kafka-topics",
          "kafka-verifiable-consumer", "kafka-verifiable-producer", "ksql", "zookeeper-shell"]
@@ -14,12 +14,12 @@
         ["kafka-acls", "kafka-avro-console-consumer", "kafka-avro-console-producer", "kafka-broker-api-versions",
          "kafka-configs", "kafka-console-consumer", "kafka-console-producer", "kafka-consumer-groups",
          "kafka-delete-records", "kafka-dump-log", "kafka-log-dirs", "kafka-topics", "kafka-verifiable-consumer",
-         "kafka-verifiable-producer"]
+         "kafka-verifiable-producer", "kafka-replica-verification", ]
     ),
     (
         "k",
         ["ksql", "kafka-acls", "kafka-avro-console-consumer", "kafka-avro-console-producer",
-         "kafka-broker-api-versions",
+         "kafka-broker-api-versions", "kafka-replica-verification",
          "kafka-configs", "kafka-console-consumer", "kafka-console-producer", "kafka-consumer-groups",
          "kafka-delete-records", "kafka-dump-log", "kafka-log-dirs", "kafka-topics", "kafka-verifiable-consumer",
          "kafka-verifiable-producer", "zookeeper-shell"]

tests/test_executor.py

@@ -113,6 +113,22 @@
         "kafka-consumer-groups --list --bootstrap-server test:9092",
         "kafka-consumer-groups --list --bootstrap-server test:9092"
     ),
+    (
+        "kafka-replica-verification",
+        "kafka-replica-verification --broker-list localhost:9092"
+    ),
+    (
+        "kafka-replica-verification --fetch-size 1024",
+        "kafka-replica-verification --fetch-size 1024 --broker-list localhost:9092"
+    ),
+    (
+        "kafka-replica-verification --broker-list broker1:9092,broker2:9093",
+        "kafka-replica-verification --broker-list broker1:9092,broker2:9093"
+    ),
+    (
+        "kafka-replica-verification --broker-list broker1:9092,broker2:9093 --fetch-size 1024",
+        "kafka-replica-verification --broker-list broker1:9092,broker2:9093 --fetch-size 1024"
+    ),
     (
         "kafka-broker-api-versions",
         "kafka-broker-api-versions --bootstrap-server localhost:9092"