Confluent Schema Registry Api Laravel Package

Getting Started

Minimal Setup

1. Install the package:

   composer require flix-tech/confluent-schema-registry-api
   

2. Configure HTTP client (e.g., in config/services.php):

   'schema-registry' => [
       'url' => env('SCHEMA_REGISTRY_URL', 'http://localhost:8081'),
       'auth' => [
           'basic' => [
               'username' => env('SCHEMA_REGISTRY_USER'),
               'password' => env('SCHEMA_REGISTRY_PASSWORD'),
           ],
       ],
   ],
   

3. Basic usage in a Laravel service:

   use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
   use GuzzleHttp\Client;
   use FlixTech\AvroPhp\AvroSchema;
   
   class SchemaService {
       protected $registry;
   
       public function __construct() {
           $client = new Client([
               'base_uri' => config('services.schema-registry.url'),
               'auth' => config('services.schema-registry.auth.basic'),
           ]);
           $this->registry = new BlockingRegistry(
               new PromisingRegistry($client)
           );
       }
   
       public function registerSchema(string $subject, string $schemaJson): int {
           $schema = AvroSchema::parse($schemaJson);
           return $this->registry->register($subject, $schema);
       }
   }
   

First Use Case: Schema Validation in Laravel

Use the package to validate Kafka payloads against registered schemas in a Laravel request handler:

use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
use FlixTech\AvroPhp\AvroSchema;

class KafkaWebhookController extends Controller {
    protected $registry;

    public function __construct(BlockingRegistry $registry) {
        $this->registry = $registry;
    }

    public function handleWebhook(Request $request) {
        $payload = $request->json();
        $schemaId = $this->registry->schemaId('webhook-payload', AvroSchema::parse($payload));

        // Proceed with validated payload
    }
}

Implementation Patterns

1. Synchronous vs. Asynchronous Workflows

• Synchronous (BlockingRegistry): Ideal for Laravel’s synchronous routes, commands, or jobs where immediate results are needed.

  $schemaId = $registry->register('user-profile', $schema);
  

• Asynchronous (PromisingRegistry): Useful in Laravel’s queue workers or event listeners where non-blocking operations are preferred.

  $promise = $registry->register('user-profile', $schema);
  $promise->then(function ($schemaId) {
      // Handle schema ID asynchronously
  });
  

2. Caching Strategies

Leverage caching to reduce Schema Registry API calls in high-traffic Laravel applications:

use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
use FlixTech\SchemaRegistryApi\Registry\Cache\DoctrineCacheAdapter;
use Doctrine\Common\Cache\ArrayCache;

$cache = new DoctrineCacheAdapter(new ArrayCache());
$cachedRegistry = new CachedRegistry($promisingRegistry, $cache);

// Cache schema IDs by schema hash (default: md5)
$schemaId = $cachedRegistry->register('user-profile', $schema);

3. Integration with Laravel’s Service Container

Bind the registry to Laravel’s container in AppServiceProvider:

public function register() {
    $this->app->singleton(BlockingRegistry::class, function ($app) {
        $client = new Client([
            'base_uri' => config('services.schema-registry.url'),
        ]);
        return new BlockingRegistry(new PromisingRegistry($client));
    });
}

Inject the registry into controllers, commands, or jobs:

public function __construct(BlockingRegistry $registry) {
    $this->registry = $registry;
}

4. Schema Management in Migrations

Use the package to manage schema versions in Laravel migrations:

public function up() {
    $schema = AvroSchema::parse(file_get_contents('schemas/user.avsc'));
    $schemaId = $this->registry->register('user', $schema);

    Schema::create('users', function (Blueprint $table) use ($schemaId) {
        $table->id();
        $table->integer('schema_id')->default($schemaId);
        // ...
    });
}

5. Error Handling

Wrap registry operations in Laravel’s exception handling:

use FlixTech\SchemaRegistryApi\Exception\SchemaRegistryException;

try {
    $schemaId = $this->registry->register('user-profile', $schema);
} catch (SchemaRegistryException $e) {
    report($e); // Log to Laravel's error reporting
    throw new \RuntimeException('Schema registration failed', 0, $e);
}

6. Low-Level API for Custom Endpoints

Use the low-level API to build custom requests (e.g., for schema deletion):

use FlixTech\SchemaRegistryApi\Requests\Functions;

$request = Functions::deleteSchema($client, 'subject', 1);
$response = $client->send($request);

Gotchas and Tips

Pitfalls

1. Schema Registry Compatibility:

   • Ensure your Confluent Schema Registry version matches the package’s supported API endpoints. Test with the exact version you’re using in production.
   • The package assumes Avro schemas by default (via flix-tech/avro-php). For Protobuf or JSON Schema, you’ll need to pre-process schemas or extend the package.

2. Caching Quirks:

   • Cache Invalidation: The CachedRegistry does not automatically invalidate cached schemas when they’re updated. Implement a custom cache adapter or use a TTL-based strategy.
   • Schema Hash Collisions: Default md5 hashing may cause collisions for large schemas. Use sha1 or a custom hash function for critical applications:

     $cachedRegistry = new CachedRegistry($promisingRegistry, $cache, 'sha1');
     

3. Asynchronous Edge Cases:

   • Uncaught Exceptions: Promises reject with SchemaRegistryException, but uncaught exceptions in .then() callbacks will throw. Always handle rejections explicitly:

     $promise->then(function ($schemaId) {
         return $schemaId;
     })->otherwise(function (SchemaRegistryException $e) {
         // Log or rethrow
     });
     

   • Promise Leaks: Unresolved promises can leak memory. Always call .wait() or handle promises in a bounded context (e.g., Laravel jobs).

4. Authentication:

   • The package does not enforce authentication schemes. Configure Guzzle’s client with the correct auth (e.g., Basic Auth, Bearer tokens) based on your Schema Registry setup:

     $client = new Client([
         'base_uri' => config('services.schema-registry.url'),
         'auth' => ['user', 'pass'], // Basic Auth
         // 'headers' => ['Authorization' => 'Bearer token'], // Bearer token
     ]);
     

5. Schema Registry Performance:

   • Rate Limiting: Schema Registry may throttle requests. Implement exponential backoff in Laravel’s App\Exceptions\Handler:

     public function render($request, Throwable $exception) {
         if ($exception instanceof SchemaRegistryException && $exception->getCode() === 429) {
             return response()->json(['error' => 'Rate limited'], 429);
         }
         return parent::render($request, $exception);
     }
     

   • Bulk Operations: For bulk schema registrations, batch requests manually to avoid hitting rate limits.

Debugging Tips

1. Enable Guzzle Middleware: Add a logging middleware to debug HTTP requests:

   $client = new Client([
       'base_uri' => config('services.schema-registry.url'),
       'middleware' => [
           new \GuzzleHttp\Middleware::tap(function ($request) {
               \Log::debug('Schema Registry Request', [
                   'method' => $request->getMethod(),
                   'uri' => (string) $request->getUri(),
                   'body' => (string) $request->getBody(),
               ]);
           }),
       ],
   ]);
   

2. Validate Schema Registry Health: Check the registry’s health endpoint in Laravel’s routes/api.php:

   Route::get('/schema-registry/health', function () {
       $client = new Client(['base_uri' => config('services.schema-registry.url')]);
       $response = $client->get('/subjects');
       return response()->json(['status' => 'healthy']);
   });
   

3. Schema Registry Logs:

   • For self-hosted Schema Registry, check logs at /var/log/confluent/schema-registry/schema-registry.log (default path).
   • For Confluent Cloud, use the Confluent Control Center or API logs.

Extension Points

1. Custom Cache Adapters: Implement CacheAdapterInterface for custom caching backends (e.g., Redis):

   use FlixTech\SchemaRegistryApi\Registry\Cache\CacheAdapterInterface;
   use Psr\SimpleCache\