Skip to content

Commit

Permalink
move sync threshold for producers (#462)
Browse files Browse the repository at this point in the history
  • Loading branch information
mensfeld authored Oct 21, 2024
1 parent 4fc5720 commit 6a21573
Show file tree
Hide file tree
Showing 5 changed files with 23 additions and 10 deletions.
1 change: 1 addition & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@

## 0.10.4 (Unreleased)
- **[Breaking]** Drop Ruby `3.0` support according to the EOL schedule.
- [Enhancement] Extract producers tracking `sync_threshold` into an internal config
- [Fix] Toggle menu button post-turbo refresh stops working.

## 0.10.3 (2024-09-17)
Expand Down
7 changes: 6 additions & 1 deletion lib/karafka/web/config.rb
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ class Config
# Reports the metrics collected in the consumer sampler
setting :reporter, default: Tracking::Consumers::Reporter.new

# Minimum number of messages to produce to produce them in sync mode
# Minimum number of messages to produce them in sync mode
# This acts as a small back-off not to overload the system in case we would have
# extremely big number of errors and reports happening
setting :sync_threshold, default: 50
Expand All @@ -98,6 +98,11 @@ class Config
end

setting :producers do
# Minimum number of messages to produce them in sync mode
# This acts as a small back-off not to overload the system in case we would have
# extremely big number of errors happening
setting :sync_threshold, default: 25

# Reports the metrics collected in the producer sampler
setting :reporter, default: Tracking::Producers::Reporter.new

Expand Down
1 change: 1 addition & 0 deletions lib/karafka/web/contracts/config.rb
Original file line number Diff line number Diff line change
Expand Up @@ -41,6 +41,7 @@ class Config < Web::Contracts::Base
required(:reporter) { |val| !val.nil? }
required(:sampler) { |val| !val.nil? }
required(:listeners) { |val| val.is_a?(Array) }
required(:sync_threshold) { |val| val.is_a?(Integer) && val.positive? }
end
end

Expand Down
9 changes: 1 addition & 8 deletions lib/karafka/web/tracking/producers/reporter.rb
Original file line number Diff line number Diff line change
Expand Up @@ -10,13 +10,6 @@ module Producers
# because there is no expectation on immediate status updates for producers and their
# dispatch flow is always periodic based.
class Reporter < Tracking::Reporter
# Minimum number of messages to produce to produce them in sync mode
# This acts as a small back-off not to overload the system in case we would have
# extremely big number of errors happening
PRODUCE_SYNC_THRESHOLD = 25

private_constant :PRODUCE_SYNC_THRESHOLD

# This mutex is shared between tracker and samplers so there is no case where metrics
# would be collected same time tracker reports
MUTEX = Mutex.new
Expand Down Expand Up @@ -82,7 +75,7 @@ def sampler
# normal operations we should not have that many messages to dispatch and it should not
# slowdown any processing.
def produce(messages)
if messages.count >= PRODUCE_SYNC_THRESHOLD
if messages.count >= ::Karafka::Web.config.tracking.producers.sync_threshold
::Karafka::Web.producer.produce_many_sync(messages)
else
::Karafka::Web.producer.produce_many_async(messages)
Expand Down
15 changes: 14 additions & 1 deletion spec/lib/karafka/web/contracts/config_spec.rb
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,8 @@
producers: {
reporter: Object.new,
sampler: Object.new,
listeners: []
listeners: [],
sync_threshold: 10
}
},
processing: {
Expand Down Expand Up @@ -119,6 +120,18 @@
it { expect(contract.call(params)).not_to be_success }
end

context 'when producers sync_threshold is less than 0' do
before { params[:tracking][:producers][:sync_threshold] = -1 }

it { expect(contract.call(params)).not_to be_success }
end

context 'when producers sync_threshold is not an integer' do
before { params[:tracking][:producers][:sync_threshold] = 1.1 }

it { expect(contract.call(params)).not_to be_success }
end

%i[consumers producers].each do |entity|
context "when checking #{entity} scoped data" do
%i[reporter sampler].each do |field|
Expand Down

0 comments on commit 6a21573

Please sign in to comment.