CCOL-2039: Post process valid and invalid records from batch consumption

lib/deimos.rb

@@ -57,7 +57,20 @@ def schema_backend_class
     # @param namespace [String]
     # @return [Deimos::SchemaBackends::Base]
     def schema_backend(schema:, namespace:)
-      schema_backend_class.new(schema: schema, namespace: namespace)
+      if Utils::SchemaClass.use?(config.to_h)
+        # Initialize an instance of the provided schema
+        # in the event the schema class is an override, the inherited
+        # schema and namespace will be applied
+        schema_class = Utils::SchemaClass.klass(schema, namespace)
+        if schema_class.nil?
+          schema_backend_class.new(schema: schema, namespace: namespace)
+        else
+          schema_instance = schema_class.new
+          schema_backend_class.new(schema: schema_instance.schema, namespace: schema_instance.namespace)
+        end
+      else
+        schema_backend_class.new(schema: schema, namespace: namespace)
+      end
     end
 
     # @param schema [String]

lib/deimos/consumer.rb

@@ -19,17 +19,8 @@ class Consumer
     class << self
       # @return [Deimos::SchemaBackends::Base]
       def decoder
-        return @decoder if @decoder
-
-        @decoder = if Utils::SchemaClass.use?(config.to_h)
-                     # Initialize an instance of the provided schema
-                     # in the event the schema class is an override, the inherited
-                     # schema and namespace will be applied
-                     schema_class = "Schemas::#{config[:schema]}".constantize.new
-                     Deimos.schema_backend(schema: schema_class.schema, namespace: schema_class.namespace)
-                   else
-                     Deimos.schema_backend(schema: config[:schema], namespace: config[:namespace])
-                   end
+        @decoder ||= Deimos.schema_backend(schema: config[:schema],
+                                           namespace: config[:namespace])
       end
 
       # @return [Deimos::SchemaBackends::Base]

lib/deimos/tracing/datadog.rb

@@ -11,12 +11,11 @@ def initialize(config)
         raise 'Tracing config must specify service_name' if config[:service_name].nil?
 
         @service = config[:service_name]
-        @tracer = ::Datadog.respond_to?(:tracer) ? ::Datadog.tracer : ::Datadog::Tracing
       end
 
       # :nodoc:
       def start(span_name, options={})
-        span = @tracer.trace(span_name)
+        span = tracer.trace(span_name)
         span.service = @service
         span.resource = options[:resource]
         span
@@ -27,9 +26,14 @@ def finish(span)
         span.finish
       end
 
+      # :nodoc:
+      def tracer
+        @tracer ||= ::Datadog.respond_to?(:tracer) ? ::Datadog.tracer : ::Datadog::Tracing
+      end
+
       # :nodoc:
       def active_span
-        @tracer.active_span
+        tracer.active_span
       end
 
       # :nodoc:

lib/deimos/utils/schema_class.rb

@@ -25,13 +25,21 @@ def modules_for(namespace)
         def instance(payload, schema, namespace='')
           return payload if payload.is_a?(Deimos::SchemaClass::Base)
 
-          constants = modules_for(namespace) + [schema.underscore.camelize.singularize]
-          klass = constants.join('::').safe_constantize
+          klass = klass(schema, namespace)
           return payload if klass.nil? || payload.nil?
 
           klass.new(**payload.symbolize_keys)
         end
 
+        # Determine and return the SchemaClass with the provided schema and namespace
+        # @param schema [String]
+        # @param namespace [String]
+        # @return [Deimos::SchemaClass]
+        def klass(schema, namespace)
+          constants = modules_for(namespace) + [schema.underscore.camelize.singularize]
+          constants.join('::').safe_constantize
+        end
+
         # @param config [Hash] Producer or Consumer config
         # @return [Boolean]
         def use?(config)

spec/active_record_batch_consumer_spec.rb

@@ -35,33 +35,6 @@ class Widget < ActiveRecord::Base
       stub_const('MyBatchConsumer', consumer_class)
       stub_const('ConsumerTest::MyBatchConsumer', consumer_class)
       consumer_class.config[:bulk_import_id_column] = :bulk_import_id # default
-      schema_class = Class.new(Deimos::SchemaClass::Record) do
-        def schema
-          'MySchema'
-        end
-
-        def namespace
-          'com.my-namespace'
-        end
-
-        attr_accessor :test_id
-        attr_accessor :some_int
-
-        def initialize(test_id: nil,
-                       some_int: nil)
-          self.test_id = test_id
-          self.some_int = some_int
-        end
-
-        def as_json(_opts={})
-          {
-            'test_id' => @test_id,
-            'some_int' => @some_int,
-            'payload_key' => @payload_key&.as_json
-          }
-        end
-      end
-      stub_const('Schemas::MySchema', schema_class)
     end
 
     around(:each) do |ex|
@@ -101,8 +74,12 @@ def publish_batch(messages)
     describe 'consume_batch' do
       SCHEMA_CLASS_SETTINGS.each do |setting, use_schema_classes|
         context "with Schema Class consumption #{setting}" do
+
           before(:each) do
-            Deimos.configure { |config| config.schema.use_schema_classes = use_schema_classes }
+            Deimos.configure do |config|
+              config.schema.use_schema_classes = use_schema_classes
+              config.schema.generate_namespace_folders = true
+            end
           end
 
           it 'should handle an empty batch' do

spec/active_record_consumer_spec.rb

@@ -141,7 +141,10 @@ def as_json(_opts={})
       SCHEMA_CLASS_SETTINGS.each do |setting, use_schema_classes|
         context "with Schema Class consumption #{setting}" do
           before(:each) do
-            Deimos.configure { |config| config.schema.use_schema_classes = use_schema_classes }
+            Deimos.configure do |config|
+              config.schema.use_schema_classes = use_schema_classes
+              config.schema.generate_namespace_folders = true
+            end
           end
 
           it 'should receive events correctly' do

spec/active_record_producer_spec.rb

@@ -69,7 +69,10 @@ def self.post_process(batch)
     SCHEMA_CLASS_SETTINGS.each do |setting, use_schema_classes|
       context "with Schema Class consumption #{setting}" do
         before(:each) do
-          Deimos.configure { |config| config.schema.use_schema_classes = use_schema_classes }
+          Deimos.configure do |config|
+            config.schema.use_schema_classes = use_schema_classes
+            config.schema.generate_namespace_folders = true
+          end
         end
 
         it 'should send events correctly' do

spec/batch_consumer_spec.rb

@@ -16,33 +16,6 @@ def consume_batch(_payloads, _metadata)
         end
       end
       stub_const('ConsumerTest::MyBatchConsumer', consumer_class)
-      schema_class = Class.new(Deimos::SchemaClass::Record) do
-        def schema
-          'MySchema'
-        end
-
-        def namespace
-          'com.my-namespace'
-        end
-
-        attr_accessor :test_id
-        attr_accessor :some_int
-
-        def initialize(test_id: nil,
-                       some_int: nil)
-          self.test_id = test_id
-          self.some_int = some_int
-        end
-
-        def as_json(_opts={})
-          {
-            'test_id' => @test_id,
-            'some_int' => @some_int,
-            'payload_key' => @payload_key&.as_json
-          }
-        end
-      end
-      stub_const('Schemas::MySchema', schema_class)
     end
 
     let(:batch) do
@@ -67,7 +40,10 @@ def as_json(_opts={})
           end
 
           before(:each) do
-            Deimos.configure { |config| config.schema.use_schema_classes = use_schema_classes }
+            Deimos.configure do |config|
+              config.schema.use_schema_classes = use_schema_classes
+              config.schema.generate_namespace_folders = true
+            end
           end
 
           it 'should provide backwards compatibility for BatchConsumer class' do

spec/consumer_spec.rb

@@ -22,40 +22,17 @@ def consume(_payload, _metadata)
         end
       end
       stub_const('ConsumerTest::MyConsumer', consumer_class)
-      schema_class = Class.new(Deimos::SchemaClass::Record) do
-        def schema
-          'MySchema'
-        end
-
-        def namespace
-          'com.my-namespace'
-        end
-
-        attr_accessor :test_id
-        attr_accessor :some_int
-
-        def initialize(test_id: nil,
-                       some_int: nil)
-          self.test_id = test_id
-          self.some_int = some_int
-        end
-
-        def as_json(_opts={})
-          {
-            'test_id' => @test_id,
-            'some_int' => @some_int,
-            'payload_key' => @payload_key&.as_json
-          }
-        end
-      end
-      stub_const('Schemas::MySchema', schema_class)
     end
 
     describe 'consume' do
       SCHEMA_CLASS_SETTINGS.each do |setting, use_schema_classes|
         context "with Schema Class consumption #{setting}" do
+
           before(:each) do
-            Deimos.configure { |config| config.schema.use_schema_classes = use_schema_classes }
+            Deimos.configure do |config|
+              config.schema.use_schema_classes = use_schema_classes
+              config.schema.generate_namespace_folders = true
+            end
           end
 
           it 'should consume a message' do
@@ -155,6 +132,41 @@ def as_json(_opts={})
           end
         end
       end
+
+      context 'with overriden schema classes' do
+
+        before(:each) do
+          Deimos.configure do |config|
+            config.schema.use_schema_classes = true
+            config.schema.generate_namespace_folders = true
+          end
+        end
+
+        prepend_before(:each) do
+          consumer_class = Class.new(described_class) do
+            schema 'MyUpdatedSchema'
+            namespace 'com.my-namespace'
+            key_config field: 'test_id'
+
+            # :nodoc:
+            def consume(_payload, _metadata)
+              raise 'This should not be called unless call_original is set'
+            end
+          end
+          stub_const('ConsumerTest::MyConsumer', consumer_class)
+        end
+
+        it 'should consume messages' do
+          test_consume_message('my_consume_topic',
+                               { 'test_id' => 'foo',
+                                 'some_int' => 1 }) do |payload, _metadata|
+            expect(payload['test_id']).to eq('foo')
+            expect(payload['some_int']).to eq(1)
+            expect(payload['super_int']).to eq(9000)
+                                 end
+        end
+
+      end
     end
 
     describe 'decode_key' do

spec/schemas/my_namespace/my_schema_updated.rb

@@ -0,0 +1,18 @@
+# frozen_string_literal: true
+
+# This file is autogenerated by Deimos, Do NOT modify
+module Schemas; module MyNamespace
+  ### Primary Schema Class ###
+  # Autogenerated Schema for Record at com.my-namespace.MySchema
+  class MyUpdatedSchema < Schemas::MyNamespace::MySchema
+
+    attr_accessor :super_int
+
+    def initialize(test_id: nil,
+                   some_int: nil)
+      super
+      self.super_int = some_int.nil? ? 10 : some_int * 9000
+    end
+  end
+end
+end