feat: Raise BatchNotSupportedError for batch content in KafkaBinding

Add BatchNotSupportedError to the error hierarchy and raise it from
KafkaBinding#decode_event when given an application/cloudevents-batch
content type, since the Kafka protocol binding does not support batch
content mode per the spec.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Daniel Azuma <dazuma@gmail.com>

Author: dazuma

lib/cloud_events/errors.rb

@@ -51,6 +51,14 @@ class SpecVersionError < CloudEventsError
   class AttributeError < CloudEventsError
   end
 
+  ##
+  # An error raised when a protocol binding that does not support batch
+  # content mode receives a batch. For example, the Kafka protocol binding
+  # does not support batches per the CloudEvents spec.
+  #
+  class BatchNotSupportedError < CloudEventsError
+  end
+
   ##
   # Alias of UnsupportedFormatError, for backward compatibility.
   #

lib/cloud_events/kafka_binding.rb

@@ -187,15 +187,7 @@ def decode_event(message, allow_opaque: false, reverse_key_mapper: :NOT_SET, **f
       content_type_string = headers["content-type"]
       content_type = ContentType.new(content_type_string) if content_type_string
 
-      event = if content_type&.media_type == "application" && content_type.subtype_base == "cloudevents"
-                decode_structured_content(message, content_type, allow_opaque, **format_args)
-              elsif headers.key?("ce_specversion")
-                decode_binary_content(message, content_type, **format_args)
-              end
-      unless event
-        ct_desc = content_type_string ? content_type_string.inspect : "not present"
-        raise(NotCloudEventError, "content-type is #{ct_desc}, and ce_specversion header is not present")
-      end
+      event = decode_content(message, headers, content_type, content_type_string, allow_opaque, **format_args)
       apply_reverse_key_mapper(event, message[:key], reverse_key_mapper)
     end
 
@@ -229,6 +221,22 @@ def encode_event(event, structured_format: false, key_mapper: :NOT_SET, **format
 
     private
 
+    # Detect content mode, reject batches, and dispatch to the appropriate
+    # binary or structured decoder. Raises if the message is not a CloudEvent.
+    def decode_content(message, headers, content_type, content_type_string, allow_opaque, **format_args)
+      if content_type&.media_type == "application" && content_type.subtype_base == "cloudevents-batch"
+        raise(BatchNotSupportedError, "Kafka protocol binding does not support batch content mode")
+      end
+      if content_type&.media_type == "application" && content_type.subtype_base == "cloudevents"
+        return decode_structured_content(message, content_type, allow_opaque, **format_args)
+      end
+      if headers.key?("ce_specversion")
+        return decode_binary_content(message, content_type, **format_args)
+      end
+      ct_desc = content_type_string ? content_type_string.inspect : "not present"
+      raise(NotCloudEventError, "content-type is #{ct_desc}, and ce_specversion header is not present")
+    end
+
     # Decode a single event from binary content mode. Reads ce_* headers as
     # attributes and the message value as event data.
     def decode_binary_content(message, content_type, **format_args)

test/test_kafka_binding.rb

@@ -274,6 +274,17 @@
         kafka_binding.decode_event(message, reverse_key_mapper: nil)
       end
     end
+
+    it "raises BatchNotSupportedError for batch content type" do
+      message = {
+        key: nil,
+        value: "[{}]",
+        headers: { "content-type" => "application/cloudevents-batch+json" },
+      }
+      assert_raises CloudEvents::BatchNotSupportedError do
+        kafka_binding.decode_event(message, reverse_key_mapper: nil)
+      end
+    end
   end
 
   describe "decode_event structured mode" do