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 (Laravel-specific):

   use GuzzleHttp\Client;
   use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
   
   $client = new Client([
       'base_uri' => config('services.schema_registry.url'),
       'timeout'  => 5.0,
   ]);
   
   $registry = new BlockingRegistry(
       new PromisingRegistry($client)
   );
   

3. First Use Case: Register a schema for a Kafka topic subject:

   use FlixTech\AvroPhp\AvroSchema;
   
   $schema = AvroSchema::parse(file_get_contents('path/to/schema.avsc'));
   $schemaId = $registry->register('user-events-value', $schema);
   

Where to Look First

• Core Classes:
  • BlockingRegistry (synchronous, Laravel-friendly)
  • PromisingRegistry (asynchronous, for event-driven workflows)
  • CachedRegistry (for performance optimization)
• Laravel Integration:
  • Bind the registry to the container in AppServiceProvider:

    $this->app->singleton('schema.registry', fn() => new BlockingRegistry(
        new PromisingRegistry(new Client(['base_uri' => config('services.schema_registry.url')]))
    ));
    

  • Use dependency injection in controllers/services:

    public function __construct(private BlockingRegistry $registry) {}
    

Implementation Patterns

Workflows

1. Schema Registration & Validation

• Pattern: Use BlockingRegistry in Laravel’s HTTP layer or service classes.
• Example:

  public function storeSchema(Request $request, BlockingRegistry $registry) {
      $schema = AvroSchema::parse($request->input('schema'));
      $schemaId = $registry->register('orders-value', $schema);
      return response()->json(['schema_id' => $schemaId]);
  }
  

• Laravel-Specific: Validate input schemas using Laravel’s FormRequest:

  public function rules() {
      return ['schema' => 'required|json'];
  }
  

2. Schema Retrieval for Kafka Producers/Consumers

• Pattern: Cache schemas to avoid repeated API calls.
• Example:

  public function getSchemaForTopic(string $subject, BlockingRegistry $registry) {
      return cache()->remember("schema:{$subject}", now()->addHours(1), function() use ($subject, $registry) {
          return $registry->latestSchema($subject);
      });
  }
  

3. Schema Evolution & Compatibility Checks

• Pattern: Use schemaCompatibility() to enforce backward/forward compatibility.
• Example:

  public function updateSchema(Request $request, BlockingRegistry $registry) {
      $currentSchema = $registry->latestSchema('user-events-value');
      $newSchema = AvroSchema::parse($request->input('schema'));
  
      if (!$registry->schemaCompatibility('user-events-value', $currentSchema, $newSchema, 'BACKWARD')) {
          throw new \RuntimeException('Schema is not backward compatible!');
      }
  
      return $registry->register('user-events-value', $newSchema);
  }
  

4. Asynchronous Processing (Queues/Jobs)

• Pattern: Use PromisingRegistry in Laravel queues for non-blocking schema operations.
• Example:

  public function handle(RegisterSchemaJob $job, PromisingRegistry $registry) {
      $promise = $registry->register($job->subject, $job->schema);
      $promise->then(
          fn($schemaId) => $job->resolve($schemaId),
          fn($exception) => $job->fail($exception)
      );
  }
  

Integration Tips

• Laravel Config: Add to config/services.php:

  'schema_registry' => [
      'url' => env('SCHEMA_REGISTRY_URL', 'http://localhost:8081'),
      'cache' => env('SCHEMA_REGISTRY_CACHE', 'array'), // 'array', 'doctrine', or 'simple_cache'
  ],
  

• Service Provider: Bind cache adapters dynamically:

  $this->app->bind('schema.registry.cache', function($app) {
      $config = config('services.schema_registry.cache');
      return match($config) {
          'doctrine' => new DoctrineCacheAdapter($app->make(\Doctrine\Common\Cache\CacheProvider::class)),
          'simple_cache' => new SimpleCacheAdapter($app->make(\Psr\SimpleCache\CacheInterface::class)),
          default => new ArrayCacheAdapter(),
      };
  });
  

• Testing: Use Laravel’s HTTP testing to mock the registry:

  $response = $this->post('/schemas', ['schema' => '{"type": "string"}']);
  $response->assertJson(['schema_id' => 1]);
  

Gotchas and Tips

Pitfalls

1. Schema Hash Collisions:
   • The default md5 hash for caching schemas may collide. Use a stronger hash (e.g., sha256) if schemas are large or complex:

     $registry = new CachedRegistry($promisingRegistry, new AvroObjectCacheAdapter(), fn(AvroSchema $schema) => hash('sha256', (string)$schema));
     

2. Blocking Calls in Synchronous Contexts:
   • BlockingRegistry uses wait() internally, which can block Laravel’s request lifecycle. Prefer PromisingRegistry for long-running operations (e.g., background jobs).
3. Avro Schema Parsing Errors:
   • Invalid Avro schemas throw AvroException. Validate schemas before registration:

     try {
         $schema = AvroSchema::parse($rawSchema);
     } catch (\Exception $e) {
         throw new \InvalidArgumentException('Invalid Avro schema', 0, $e);
     }
     

4. Caching Invalidation:
   • CachedRegistry does not auto-invalidate cache on schema updates. Manually clear cache when needed:

     cache()->forget("schema:{$subject}");
     

Debugging

• Enable Guzzle Middleware: Add logging to Guzzle requests for debugging:

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

• Schema Registry Logs: Check Confluent Schema Registry logs for 4xx/5xx errors (e.g., schema.registry.log in Docker setups).

Extension Points

1. Custom Cache Adapters: Implement CacheAdapterInterface for Laravel’s cache:

   use FlixTech\SchemaRegistryApi\Registry\Cache\CacheAdapterInterface;
   use Illuminate\Contracts\Cache\Store;
   
   class LaravelCacheAdapter implements CacheAdapterInterface {
       public function __construct(private Store $cache) {}
   
       public function fetch($key) { /* ... */ }
       public function save($key, $value, $ttl) { /* ... */ }
       public function delete($key) { /* ... */ }
   }
   

2. Schema Validation Middleware: Create a Laravel middleware to validate schemas before registration:

   public function handle($request, Closure $next) {
       $schema = AvroSchema::parse($request->input('schema'));
       if (!$schema->isValid()) {
           abort(422, 'Invalid schema');
       }
       return $next($request);
   }
   

3. Event Listeners for Schema Changes: Dispatch Laravel events when schemas are registered/updated:

   $registry->register($subject, $schema)->then(
       fn($schemaId) => event(new SchemaRegistered($subject, $schemaId))
   );
   

Laravel-Specific Quirks

• Queue Timeouts: PromisingRegistry operations in queues may hit Laravel’s queue.worker.timeout (default: 60s). Increase if needed:

  'queue' => [
      'worker' => [
          'timeout' => 300, // 5 minutes
      ],
  ],
  

• Service Container Binding: Avoid circular dependencies by binding BlockingRegistry to PromisingRegistry explicitly:

  $this->app->bind(BlockingRegistry::class, fn($app) => new BlockingRegistry(
      $app->make(PromisingRegistry::class)
  ));
  

• Testing: Mock the registry in tests to avoid real API calls:

  $mockRegistry = Mockery::mock(BlockingRegistry::class);
  $mockRegistry->shouldReceive('register')->andReturn