CCOL-2039: CR fixes & configuration spec

lib/deimos/active_record_consume/batch_consumption.rb

@@ -178,7 +178,7 @@ def upsert_records(messages)
                                   key_col_proc: key_col_proc,
                                   col_proc: col_proc,
                                   replace_associations: self.class.replace_associations,
-                                  batch_id_generator: self.class.bulk_import_id_generator)
+                                  bulk_import_id_generator: self.class.bulk_import_id_generator)
         [updater.mass_update(record_list), invalid]
       end
 
@@ -220,7 +220,8 @@ def build_records(messages)
 
           BatchRecord.new(klass: @klass,
                           attributes: attrs,
-                          bulk_import_column: col)
+                          bulk_import_column: col,
+                          bulk_import_id_generator: self.class.bulk_import_id_generator)
         end
         BatchRecordList.new(records.compact)
       end

lib/deimos/active_record_consume/batch_record.rb

@@ -22,12 +22,12 @@ class BatchRecord
       # @param klass [Class < ActiveRecord::Base]
       # @param attributes [Hash] the full attribute list, including associations.
       # @param bulk_import_column [String]
-      # @param bulk_id_generator [Proc]
-      def initialize(klass:, attributes:, bulk_import_column: nil, bulk_id_generator: nil)
+      # @param bulk_import_id_generator [Proc]
+      def initialize(klass:, attributes:, bulk_import_column: nil, bulk_import_id_generator: nil)
         @klass = klass
         if bulk_import_column
           self.bulk_import_column = bulk_import_column
-          self.bulk_import_id = bulk_id_generator&.call
+          self.bulk_import_id = bulk_import_id_generator&.call
           attributes[bulk_import_column] = bulk_import_id
         end
         attributes = attributes.with_indifferent_access

lib/deimos/active_record_consume/mass_updater.rb

@@ -19,10 +19,10 @@ def default_cols(klass)
       # @param key_col_proc [Proc<Class < ActiveRecord::Base>]
       # @param col_proc [Proc<Class < ActiveRecord::Base>]
       # @param replace_associations [Boolean]
-      def initialize(klass, key_col_proc: nil, col_proc: nil, replace_associations: true, batch_id_generator: nil)
+      def initialize(klass, key_col_proc: nil, col_proc: nil, replace_associations: true, bulk_import_id_generator: nil)
         @klass = klass
         @replace_associations = replace_associations
-        @batch_id_generator = batch_id_generator || proc { SecureRandom.uuid }
+        @bulk_import_id_generator = bulk_import_id_generator || proc { SecureRandom.uuid }
 
         @key_cols = {}
         @key_col_proc = key_col_proc
@@ -70,7 +70,7 @@ def save_records_to_database(record_list)
       def import_associations(record_list)
         record_list.fill_primary_keys!
 
-        import_id = @replace_associations ? @batch_id_generator.call : nil
+        import_id = @replace_associations ? @bulk_import_id_generator.call : nil
         record_list.associations.each do |assoc|
           sub_records = record_list.map { |r| r.sub_records(assoc.name, import_id) }.flatten
           next unless sub_records.any?

spec/config/configuration_spec.rb

@@ -282,4 +282,40 @@ def consume
         }
       )
   end
+
+  it 'should override global configurations' do
+    described_class.configure do
+      consumers.bulk_import_id_generator proc { 'global' }
+      consumers.replace_associations true
+
+      consumer do
+        class_name 'MyConfigConsumer'
+        schema 'blah'
+        topic 'blah'
+        group_id 'myconsumerid'
+        bulk_import_id_generator proc { 'consumer' }
+        replace_associations false
+      end
+
+      consumer do
+        class_name 'MyConfigConsumer2'
+        schema 'blah'
+        topic 'blah'
+        group_id 'myconsumerid'
+      end
+    end
+
+    consumers = described_class.config.consumers
+    expect(consumers.replace_associations).to eq(true)
+    expect(consumers.bulk_import_id_generator.call).to eq('global')
+
+    custom = MyConfigConsumer.config
+    expect(custom[:replace_associations]).to eq(false)
+    expect(custom[:bulk_import_id_generator].call).to eq('consumer')
+
+    default = MyConfigConsumer2.config
+    expect(default[:replace_associations]).to eq(true)
+    expect(default[:bulk_import_id_generator].call).to eq('global')
+
+  end
 end