feat(CDA-1565): Add command for changing compatibility level

README.md

@@ -6,7 +6,7 @@ composer require jobcloud/php-console-kafka-schema-registry
 ```
 
 ## Requirements
-- php: >= 7.4
+- php: ^8.0
 
 ## Register commands
 You can register each command separately like this:

docker/docker-compose.ci.yml

@@ -1,4 +1,3 @@
-version: '3.2'
 services:
   php:
     build:

docker/docker-compose.yml

@@ -1,4 +1,3 @@
-version: '3.2'
 services:
   php:
     build:

src/Command/SetAllSchemasCompatibilityModeCommand.php

@@ -0,0 +1,176 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Jobcloud\SchemaConsole\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class SetAllSchemasCompatibilityModeCommand extends AbstractSchemaCommand
+{
+    protected function configure(): void
+    {
+        $this
+            ->setName('kafka-schema-registry:set:compatibility:mode:all')
+            ->setDescription('Set compatibility modes for multiple schemas from a JSON configuration file')
+            ->setHelp($this->getHelpText())
+            ->addArgument(
+                'configFile',
+                InputArgument::REQUIRED,
+                'Path to JSON configuration file containing schema-compatibility mappings'
+            );
+    }
+
+    private function getHelpText(): string
+    {
+        return <<<'HELP'
+Set compatibility modes for multiple schemas based on configuration in a JSON file.
+
+JSON File Format:
+The configuration file must be a valid JSON array with the following structure:
+
+[
+  {
+    "schemaName": "schema-subject-name",
+    "compatibilityLevel": "COMPATIBILITY_LEVEL"
+  }
+]
+
+Required Fields:
+- schemaName: The subject name of the schema in the registry
+- compatibilityLevel: One of the following compatibility levels:
+  * NONE
+  * BACKWARD
+  * BACKWARD_TRANSITIVE
+  * FORWARD
+  * FORWARD_TRANSITIVE
+  * FULL
+  * FULL_TRANSITIVE
+
+Example:
+[
+  {
+    "schemaName": "user-events",
+    "compatibilityLevel": "BACKWARD_TRANSITIVE"
+  },
+  {
+    "schemaName": "order-events",
+    "compatibilityLevel": "FORWARD"
+  },
+  {
+    "schemaName": "payment-events",
+    "compatibilityLevel": "FULL"
+  }
+]
+
+The command will process each schema in the order specified and provide feedback
+for each operation. If any schema update fails, the command will continue processing
+the remaining schemas and return a non-zero exit code at the end.
+HELP;
+    }
+
+    public function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $configFilePath = (string) $input->getArgument('configFile');
+
+        $config = $this->loadConfigFile($configFilePath, $output);
+        if (null === $config) {
+            return Command::FAILURE;
+        }
+
+        $totalSchemas = count($config);
+        $successCount = 0;
+        $failureCount = 0;
+
+        $output->writeln(sprintf('Processing %d schema compatibility configurations...', $totalSchemas));
+
+        foreach ($config as $index => $schemaConfig) {
+            if (false === $this->isValidSchemaConfig($schemaConfig)) {
+                $output->writeln(
+                    sprintf('Invalid configuration at index %d: missing schemaName or compatibilityLevel', $index)
+                );
+                $failureCount++;
+
+                continue;
+            }
+
+            $schemaName = $schemaConfig['schemaName'];
+            $compatibilityLevel = $schemaConfig['compatibilityLevel'];
+
+            $output->write(
+                sprintf('Setting compatibility mode for schema "%s" to "%s"... ', $schemaName, $compatibilityLevel)
+            );
+
+            if ($this->setSchemaCompatibility($schemaName, $compatibilityLevel, $output)) {
+                $successCount++;
+
+                continue;
+            }
+
+            $failureCount++;
+        }
+
+        $this->outputSummary($output, $totalSchemas, $successCount, $failureCount);
+
+        return $failureCount > Command::SUCCESS ? Command::FAILURE : Command::SUCCESS;
+    }
+
+    /**
+     * @return array<int, array<string, string>>|null
+     */
+    private function loadConfigFile(string $configFilePath, OutputInterface $output): ?array
+    {
+        $jsonContent = @file_get_contents($configFilePath);
+        if (false === $jsonContent) {
+            $output->writeln(sprintf('Could not read configuration file: %s', $configFilePath));
+
+            return null;
+        }
+
+        $config = json_decode($jsonContent, true);
+        if (null === $config || false === is_array($config) || false === array_is_list($config)) {
+            $output->writeln('Configuration file must contain a JSON array of schema configurations');
+
+            return null;
+        }
+
+        return $config;
+    }
+
+    private function isValidSchemaConfig(mixed $schemaConfig): bool
+    {
+        return is_array($schemaConfig)
+            && isset($schemaConfig['schemaName'])
+            && isset($schemaConfig['compatibilityLevel']);
+    }
+
+    private function setSchemaCompatibility(
+        string $schemaName,
+        string $compatibilityLevel,
+        OutputInterface $output
+    ): bool {
+        try {
+            $this->schemaRegistryApi->setSubjectCompatibilityLevel($schemaName, $compatibilityLevel);
+        } catch (\Exception $e) {
+            $output->writeln(sprintf('<error>FAILED: %s</error>', $e->getMessage()));
+
+            return false;
+        }
+
+        $output->writeln('<info>SUCCESS</info>');
+
+        return true;
+    }
+
+    private function outputSummary(OutputInterface $output, int $total, int $success, int $failure): void
+    {
+        $output->writeln('');
+        $output->writeln('=== Summary ===');
+        $output->writeln(sprintf('Total schemas processed: %d', $total));
+        $output->writeln(sprintf('Successful updates: %d', $success));
+        $output->writeln(sprintf('Failed updates: %d', $failure));
+    }
+}

src/Command/SetCompatibilityModeForSchemaCommand.php

@@ -0,0 +1,43 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Jobcloud\SchemaConsole\Command;
+
+use Symfony\Component\Console\Command\Command;
+use Symfony\Component\Console\Input\InputArgument;
+use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Output\OutputInterface;
+
+class SetCompatibilityModeForSchemaCommand extends AbstractSchemaCommand
+{
+    protected function configure(): void
+    {
+        $this
+            ->setName('kafka-schema-registry:set:schema:compatibility:mode')
+            ->setDescription('Set the compatibility mode for a given schema')
+            ->setHelp('Set the compatibility mode for a given schema')
+            ->addArgument('schemaName', InputArgument::REQUIRED, 'Name of the schema')
+            ->addArgument('compatibilityLevel', InputArgument::REQUIRED, 'Compatibility level to set');
+    }
+
+    public function execute(InputInterface $input, OutputInterface $output): int
+    {
+        $schemaName = (string) $input->getArgument('schemaName');
+        $compatibilityLevel = (string) $input->getArgument('compatibilityLevel');
+
+        try {
+            $this->schemaRegistryApi->setSubjectCompatibilityLevel($schemaName, $compatibilityLevel);
+        } catch (\Exception $e) {
+            $output->writeln(
+                sprintf('Could not change compatibility mode for schema %s: %s', $schemaName, $e->getMessage())
+            );
+
+            return Command::FAILURE;
+        }
+
+        $output->writeln(sprintf('Successfully changed compatibility mode for schema: %s', $schemaName));
+
+        return Command::SUCCESS;
+    }
+}

src/ServiceProvider/CommandServiceProvider.php

@@ -21,6 +21,8 @@
 use Jobcloud\SchemaConsole\Command\ListVersionsForSchemaCommand;
 use Jobcloud\SchemaConsole\Command\RegisterChangedSchemasCommand;
 use Jobcloud\SchemaConsole\Command\RegisterSchemaVersionCommand;
+use Jobcloud\SchemaConsole\Command\SetAllSchemasCompatibilityModeCommand;
+use Jobcloud\SchemaConsole\Command\SetCompatibilityModeForSchemaCommand;
 use Jobcloud\SchemaConsole\Command\SetImportModeCommand;
 use Jobcloud\SchemaConsole\Command\SetReadOnlyModeCommand;
 use Jobcloud\SchemaConsole\Command\SetReadWriteModeCommand;
@@ -51,6 +53,8 @@ public function register(Container $container)
                 new GetCompatibilityModeCommand($schemaRegistryApi),
                 new CheckAllSchemasCompatibilityCommand($schemaRegistryApi),
                 new GetCompatibilityModeForSchemaCommand($schemaRegistryApi),
+                new SetAllSchemasCompatibilityModeCommand($schemaRegistryApi),
+                new SetCompatibilityModeForSchemaCommand($schemaRegistryApi),
                 new GetLatestSchemaCommand($schemaRegistryApi),
                 new GetSchemaByVersionCommand($schemaRegistryApi),
                 new ListAllSchemasCommand($schemaRegistryApi),