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

lionelpereira · 2023-11-30T20:11:41Z

Pull Request Template

Description

should_consume? method accepts validation of associations
Return valid and invalid records saved during consumption for further processing in post_process method

Fixes # (issue)

Type of change

Please delete options that are not relevant.

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
This change requires a documentation update

How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration

Test A
Test B

Checklist:

My code follows the style guidelines of this project
I have performed a self-review of my own code
I have commented my code, particularly in hard-to-understand areas
I have made corresponding changes to the documentation
I have added a line in the CHANGELOG describing this change, under the UNRELEASED heading
My changes generate no new warnings
I have added tests that prove my fix is effective or that my feature works
New and existing unit tests pass locally with my changes
Any dependent changes have been merged and published in downstream modules

dorner · 2023-12-01T18:18:53Z

lib/deimos/active_record_consume/batch_consumption.rb

-          slice(messages).
-          each(&method(:update_database))
+        valid_records = []
+        invalid_records = []


Can we rename one of these variables? They represent two different things (BatchRecord and ActiveRecord).

dorner · 2023-12-01T18:19:28Z

lib/deimos/active_record_consume/batch_consumption.rb

+                                             upserted.each_slice(max_db_batch_size) do |group|
+                                               valid, invalid = upsert_records(group)
+                                               valid_upserts.push(*valid)
+                                               invalid_upserts.push(*invalid)


Simpler to do invalid_upserts.concat(invalid) instead of .push(*invalid). Same above/below.

dorner · 2023-12-01T18:19:51Z

lib/deimos/active_record_consume/batch_consumption.rb

+                                               invalid_upserts.push(*invalid)
+                                             end
+                                             valid_upserts.compact!
+                                             invalid_upserts.compact!


Is there a reason you're compacting inside each iteration instead of after all iterations are done?

lib/deimos/active_record_consume/batch_consumption.rb

dorner · 2023-12-01T18:22:57Z

lib/deimos/active_record_consume/batch_record_list.rb

+      # @param block [Proc]
+      # @return [Array<BatchRecord>]
+      def partition!(method=nil, &block)
+        valid, invalid = if method.nil?


This should never be nil since it's defined on the base class, no?

Had a case where we had to override filter_records like so

ex:

def filter_records(records) some_filter_query = filter_query(records) records.partition! do |r| some_filter_query.exclude?(r.record.id) && valid_batch_record?(r.record, r.associations) end end

Could this be done in a proc?

Not sure... I think I'd need to see this. It feels like you're diving too deep into Deimos guts to get this working - it feels wrong for calling code to have to call so many internal methods.

dorner · 2023-12-01T18:25:03Z

spec/active_record_batch_consumer_spec.rb

+        end
+      end
+
+      context 'with post processing' do


I'm confused - what's the difference between describe 'post processing' and context 'with post processing'?

This should be with compacted messages

lib/deimos/active_record_consume/batch_consumption.rb

dorner

I still don't fully understand the use case here. We seem to be adding a lot of functionality for a very laser-specific use case.

lib/deimos/active_record_consume/batch_consumption.rb

…ollection

dorner · 2023-12-04T20:24:03Z

lib/deimos/active_record_consume/batch_consumption.rb

@@ -41,9 +42,17 @@ def consume_batch(payloads, metadata)
              uncompacted_update(messages)
            end
          end
+          process_valid_records(@valid_active_records)


Is there a reason this also can't be a notification?

This could also be a notification

dorner · 2023-12-04T20:24:21Z

lib/deimos/active_record_consume/batch_consumption.rb

      # @return [Boolean]
-      def should_consume?(_record)
+      def should_consume?(_batch_record)


Why did this need to change? Did it actually change?

It did change types from ActiveRecord::Base to BatchRecord. This is in order to validate associations that are not yet saved

lib/deimos/active_record_consume/batch_consumption.rb

dorner · 2023-12-04T20:29:57Z

lib/deimos/instrumentation.rb

+
+  ActiveSupport::Notifications.subscribe('batch_consumption.invalid_records') do |*args|
+    payload = ActiveSupport::Notifications::Event.new(*args).payload
+    payload[:consumer].process_invalid_records(payload[:records])


Calling code should be able to do this themselves rather than us having a method that we call explicitly.

dorner · 2023-12-05T14:58:42Z

lib/deimos/active_record_consume/batch_consumption.rb

@@ -167,7 +178,13 @@ def upsert_records(messages)
                                  col_proc: col_proc,
                                  replace_associations: self.class.replace_associations,
                                  bulk_import_id_generator: self.class.bulk_import_id_generator)
-        updater.mass_update(record_list)
+        @valid_active_records.concat(updater.mass_update(record_list))


Can we get this working without the instance variable? I.e. return the updated records from mass_update to the calling code and have the concatenation happen there?

lib/deimos/active_record_consume/batch_consumption.rb

spec/active_record_batch_consumer_spec.rb

dorner · 2023-12-05T19:02:07Z

lib/deimos/active_record_consume/batch_consumption.rb

-            else
-              uncompacted_update(messages)
-            end
+        @tags = %W(topic:#{metadata[:topic]})


I think it's weird to pass metric tags so far down into the workflow. Can we do something like

Deimos.config.tracer.active_span.set_tag("topic", metadata[:topic]) ... Deimos.config.tracer.active_span.get_tag("topic")

We'd have to add a get_tag method to the provider class but it feels a lot better because we can separate out the tracing/metric logic from the actual business logic.

dorner · 2023-12-05T19:02:33Z

lib/deimos/active_record_consume/batch_consumption.rb

@@ -155,19 +151,35 @@ def update_database(messages)
      # @return [void]
      def upsert_records(messages)
        record_list = build_records(messages)
-        record_list.filter!(self.method(:should_consume?).to_proc)
+        invalid = filter_records(record_list)
+        unless invalid.blank?


Better to use positive conditions: if invalid.any?

dorner · 2023-12-06T14:50:44Z

lib/deimos/active_record_consume/batch_consumption.rb


-        Deimos.instrument('ar_consumer.consume_batch', @tags) do
+        Deimos.instrument('ar_consumer.consume_batch',
+                          Deimos.config.tracer.active_span.get_tag('topic')) do


lol you don't need to do get_tag here, you already have the topic.

dorner · 2023-12-06T14:52:59Z

CHANGELOG.md

@@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Feature: Add individual configuration for custom `bulk_import_id_generator` proc per consumer
 - Feature: Add global `replace_assocations` value for  for all consumers
 - Feature: Add individual `replace_assocations` value for for individual consumers
+- Feature: `should_consume?` method accepts BatchRecord associations
+- Feature: Reintroduce `filter_records` for bulk filtering of records prior to insertion
+- Feature: Return valid and invalid records saved during consumption for further processing in `batch_consumption.valid_records` and `batch_consumption.invalid_records` ActiveSupport Notifications


This should be added to the README.

dorner · 2023-12-06T14:53:35Z

lib/deimos/tracing/provider.rb

+      # @param tag [String]
+      def get_tag(tag)
+        raise NotImplementedError
+      end


We need to implement this for datadog.

…oder

dorner · 2024-01-03T15:01:22Z

lib/deimos/consumer.rb

+                     # 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)


Shouldn't this be moved into the schema_backend method rather than just used here?

dorner · 2024-01-03T15:02:10Z

lib/deimos/tracing/datadog.rb

@@ -11,15 +11,12 @@ 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


This feels more like a method than an instance variable. It isn't actually state that needs to be stored.

dorner · 2024-01-03T15:03:16Z

spec/consumer_spec.rb

+          }
+        end
+      end
+      stub_const('Schemas::MySchema', schema_class)


Is there a reason we have to add this in all four specs? If we're just trying to test overriding, can we have a dedicated spec for that? Also, why does the overriding class have to reimplement every method?

Schemas::MySchema wasn't defined anywhere. So in tests where SCHEMAS_CLASS_SETTING = on test were failing.

It should be in spec/schemas/my_namespace. If it's not showing up, maybe the settings are wrong for the tests? Or do we need to add it to spec/schemas and require it similarly to the other generated schema files?

…ckend

dorner

Looks much better! One tiny comment left :)

dorner · 2024-01-03T19:54:23Z

spec/schemas/my_namespace/my_schema_updated.rb

Filename is backwards :)

Oof. Updated lol

dorner · 2024-01-03T20:08:50Z

Time to hit the big green button! Thanks @lionelpereira !

lionelpereira requested a review from dorner December 1, 2023 14:34

dorner requested changes Dec 1, 2023

View reviewed changes

lionelpereira commented Dec 1, 2023

View reviewed changes

lib/deimos/active_record_consume/batch_consumption.rb Show resolved Hide resolved

lionelpereira requested a review from dorner December 1, 2023 22:41

dorner requested changes Dec 4, 2023

View reviewed changes

lib/deimos/active_record_consume/batch_consumption.rb Outdated Show resolved Hide resolved

lib/deimos/active_record_consume/batch_consumption.rb Outdated Show resolved Hide resolved

Lionel Pereira added 8 commits December 4, 2023 09:59

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

6553167

CCOL-2039: Lint fixes

ca6a1e3

CCOL-2039: Add optional associations for should_consume?

3b7461e

CCOL-2039: Linting changes

8d347e4

CCOL-2039: Add CHANGELOG entry && refactor valid and invalid record c…

4a688a0

…ollection

CCOL-2039: Add consume_filter when filtering records

385e58b

CCOL-2039: YARD fixes

4661b4f

CCOL-2039: Variable rename

8813fa4

lionelpereira force-pushed the CCOL-2039-post-process branch from 5ebe94c to 8813fa4 Compare December 4, 2023 15:08

CCOL-2039: Process invalid records via ActiveSupport notifications

d6181fd

dorner requested changes Dec 4, 2023

View reviewed changes

Lionel Pereira added 4 commits December 4, 2023 16:26

CCOL-2039: Process valid and invalid records via notification

2d9e8dd

CCOL-2039: Spec update

2cb9d4d

CCOL-2039: Update association spec

6a750c8

CCOL-2039: Update CHANGELOG entry

eae600e

dorner requested changes Dec 5, 2023

View reviewed changes

Lionel Pereira added 2 commits December 5, 2023 10:58

CCOL-2039: Move deadlocks to mass updater

2b8aa2d

CCOL-2039: Maintain backwards compatibility for should_consume?

ac27466

lionelpereira requested a review from dorner December 5, 2023 18:10

dorner requested changes Dec 5, 2023

View reviewed changes

CCOL-2039: Set and get topic tag

1a5b7db

lionelpereira requested a review from dorner December 5, 2023 21:06

Lionel Pereira added 3 commits December 5, 2023 16:54

CCOL-2039: Add mock span to mock tracer

ea099a3

CCOL-2039: Update mock set_tag

1ea3069

CCOL-2039: Add retries spec for mass update and deletes

1e5d3a5

dorner requested changes Dec 6, 2023

View reviewed changes

Lionel Pereira added 2 commits December 6, 2023 10:58

CCOL-2039: Update datadog span tagging

74364f6

CCOL-2039: Update README

90b5cf7

dorner previously approved these changes Dec 6, 2023

View reviewed changes

CCOL-2039: Update CONFIGURATION docs

ceff055

lionelpereira dismissed dorner’s stale review via ceff055 December 18, 2023 19:15

Lionel Pereira added 3 commits December 18, 2023 17:45

CCOL-2039: Evaluate schemaclass schema and namespace when setting dec…

b61a403

…oder

CCOL-2039: Update specs with Schema classes

97bd093

CCOL-2039: Update linting & spec cleanup

1ecee75

dorner requested changes Jan 3, 2024

View reviewed changes

Lionel Pereira added 3 commits January 3, 2024 13:40

CCOL-2039: Add SchemaClasses context, move decoder logic to schema ba…

2b4dfc2

…ckend

CCOL-2039: Use generate namespace folders in specs

1c4e6b4

CCOL-2039: Add overriden schema

4e4f774

lionelpereira requested a review from dorner January 3, 2024 19:53

dorner requested changes Jan 3, 2024

View reviewed changes

CCOL-2039: Rename updated schema class

e21bded

lionelpereira requested a review from dorner January 3, 2024 20:02

dorner approved these changes Jan 3, 2024

View reviewed changes

dorner merged commit eb628f3 into master Jan 3, 2024
5 checks passed

dorner deleted the CCOL-2039-post-process branch January 3, 2024 20:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

lionelpereira commented Nov 30, 2023

dorner Dec 1, 2023

dorner Dec 1, 2023

dorner Dec 1, 2023

dorner Dec 1, 2023

lionelpereira Dec 1, 2023

dorner Dec 1, 2023

dorner Dec 1, 2023

lionelpereira Dec 1, 2023

dorner left a comment

dorner Dec 4, 2023

lionelpereira Dec 4, 2023

dorner Dec 4, 2023

lionelpereira Dec 4, 2023

dorner Dec 4, 2023

dorner Dec 5, 2023

dorner Dec 5, 2023

dorner Dec 5, 2023

dorner Dec 6, 2023

dorner Dec 6, 2023

dorner Dec 6, 2023

dorner Jan 3, 2024

dorner Jan 3, 2024

dorner Jan 3, 2024

lionelpereira Jan 3, 2024

dorner Jan 3, 2024

dorner left a comment

dorner Jan 3, 2024

lionelpereira Jan 3, 2024

dorner commented Jan 3, 2024

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

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

Conversation

lionelpereira commented Nov 30, 2023

Pull Request Template

Description

Type of change

How Has This Been Tested?

Checklist:

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dorner left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dorner left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dorner commented Jan 3, 2024