Add IncompatibleSchemaError

Raise a more specific error when the schema from the registry is not AVRO.

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

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