From a21a6846289409c7dab31f61853b3572b7fa59d4 Mon Sep 17 00:00:00 2001
From: Kevin McCormack <kevin@mccormack.tech>
Date: Fri, 13 Dec 2024 09:52:25 -0500
Subject: [PATCH] Add IncompatibleSchemaError

Raise a more specific error when the schema from the registry is not AVRO.
---
 lib/avro_turf/messaging.rb |  5 +++++
 spec/messaging_spec.rb     | 12 ++++++++++++
 2 files changed, 17 insertions(+)

diff --git a/lib/avro_turf/messaging.rb b/lib/avro_turf/messaging.rb
index f2a0c3c..f8f65f7 100644
--- a/lib/avro_turf/messaging.rb
+++ b/lib/avro_turf/messaging.rb
@@ -10,6 +10,7 @@
 require 'avro_turf/cached_schema_registry'
 
 class AvroTurf
+  class IncompatibleSchemaError < StandardError; end
 
   # Provides a way to encode and decode messages without having to embed schemas
   # in the encoded data. Confluent's Schema Registry[1] is used to register
@@ -221,6 +222,10 @@ def decode_message(data, schema_name: nil, namespace: @namespace)
     def fetch_schema(subject:, version: 'latest')
       schema_data = @registry.subject_version(subject, version)
       schema_id = schema_data.fetch('id')
+      schema_type = schema_data['schemaType']
+      if schema_type && schema_type != "AVRO"
+        raise IncompatibleSchemaError, "The #{schema_type} schema for #{subject} is incompatible."
+      end
       schema = Avro::Schema.parse(schema_data.fetch('schema'))
       [schema, schema_id]
     end
diff --git a/spec/messaging_spec.rb b/spec/messaging_spec.rb
index e0edb3b..92b10cc 100644
--- a/spec/messaging_spec.rb
+++ b/spec/messaging_spec.rb
@@ -377,6 +377,18 @@
       it 'gets schema from registry' do
         expect(subject).to eq([schema, schema_id])
       end
+
+      context "with an incompatible schema type" do
+        let(:response) { {'id' => schema_id, 'schema' => 'blah', 'schemaType' => schema_type } }
+        let(:schema_type) { 'PROTOBUF' }
+
+        it 'raises IncompatibleSchemaError' do
+          expect { subject }.to raise_error(
+            AvroTurf::IncompatibleSchemaError,
+            "The #{schema_type} schema for #{subj} is incompatible."
+          )
+        end
+      end
     end
 
     context 'using fetch_schema_by_id' do