From ee007d4168f4e7295ef612fdd74580a6aab09657 Mon Sep 17 00:00:00 2001 From: Maciej Mensfeld Date: Thu, 28 Sep 2023 14:37:16 +0200 Subject: [PATCH] better processing crash errors (#162) --- lib/karafka/web/errors.rb | 11 ++++++- lib/karafka/web/installer.rb | 29 +++++++++++++++++-- .../web/processing/consumers/metrics.rb | 4 +++ lib/karafka/web/processing/consumers/state.rb | 4 +++ .../web/processing/consumers/metrics_spec.rb | 10 +++++++ .../web/processing/consumers/state_spec.rb | 10 +++++++ 6 files changed, 65 insertions(+), 3 deletions(-) diff --git a/lib/karafka/web/errors.rb b/lib/karafka/web/errors.rb index 324fb09b..93d1f1ef 100644 --- a/lib/karafka/web/errors.rb +++ b/lib/karafka/web/errors.rb @@ -17,9 +17,18 @@ module Processing # If you see this error, it probably means, that you did not bootstrap Web-UI correctly MissingConsumersStateError = Class.new(BaseError) - # Similar to the above. It should be created during install + # Raised when we try to materialize the state but the consumers states topic does not + # exist and we do not have a way to get the initial state. + # It differs from the above because above indicates that the topic exists but that there + # is no initial state, while this indicates, that there is no consumers states topic. + MissingConsumersStatesTopicError = Class.new(BaseError) + + # Similar to the above. It should be created during install / migration MissingConsumersMetricsError = Class.new(BaseError) + # Similar to the one related to consumers states + MissingConsumersMetricsTopicError = Class.new(BaseError) + # This error occurs when consumer running older version of the web-ui tries to materialize # states from newer versions. Karafka Web-UI provides only backwards compatibility, so # you need to have an up-to-date consumer materializing reported states. diff --git a/lib/karafka/web/installer.rb b/lib/karafka/web/installer.rb index ff52528d..b8cb70b5 100644 --- a/lib/karafka/web/installer.rb +++ b/lib/karafka/web/installer.rb @@ -18,7 +18,7 @@ def install(replication_factor: 1) puts 'Creating necessary topics and populating state data...' puts Management::CreateTopics.new.call(replication_factor) - puts + wait_for_topics Management::CreateInitialStates.new.call puts Management::ExtendBootFile.new.call @@ -36,6 +36,7 @@ def migrate(replication_factor: 1) puts 'Creating necessary topics and populating state data...' puts Management::CreateTopics.new.call(replication_factor) + wait_for_topics Management::CreateInitialStates.new.call puts puts("Migration #{green('completed')}. Have fun!") @@ -51,7 +52,7 @@ def reset(replication_factor: 1) Management::DeleteTopics.new.call puts Management::CreateTopics.new.call(replication_factor) - puts + wait_for_topics Management::CreateInitialStates.new.call puts puts("Resetting #{green('completed')}. Have fun!") @@ -74,6 +75,30 @@ def uninstall def enable! Management::Enable.new.call end + + private + + # Waits with a message, that we are waiting on topics + # This is not doing much, just waiting as there are some cases that it takes a bit of time + # for Kafka to actually propagate new topics knowledge across the cluster. We give it that + # bit of time just in case. + def wait_for_topics + puts + print 'Waiting for the topics to synchronize in the cluster' + wait(5) + puts + end + + # Waits for given number of seconds and prints `.` every second. + # @param time_in_seconds [Integer] time of wait + def wait(time_in_seconds) + time_in_seconds.times do + sleep(1) + print '.' + end + + print "\n" + end end end end diff --git a/lib/karafka/web/processing/consumers/metrics.rb b/lib/karafka/web/processing/consumers/metrics.rb index 63447ae3..b03e080f 100644 --- a/lib/karafka/web/processing/consumers/metrics.rb +++ b/lib/karafka/web/processing/consumers/metrics.rb @@ -20,6 +20,10 @@ def current! return metrics_message.payload if metrics_message raise(::Karafka::Web::Errors::Processing::MissingConsumersMetricsError) + rescue Rdkafka::RdkafkaError => e + raise(e) unless e.code == :unknown_partition + + raise(::Karafka::Web::Errors::Processing::MissingConsumersMetricsTopicError) end end end diff --git a/lib/karafka/web/processing/consumers/state.rb b/lib/karafka/web/processing/consumers/state.rb index 72d8f052..91984552 100644 --- a/lib/karafka/web/processing/consumers/state.rb +++ b/lib/karafka/web/processing/consumers/state.rb @@ -20,6 +20,10 @@ def current! return state_message.payload if state_message raise(::Karafka::Web::Errors::Processing::MissingConsumersStateError) + rescue Rdkafka::RdkafkaError => e + raise(e) unless e.code == :unknown_partition + + raise(::Karafka::Web::Errors::Processing::MissingConsumersStatesTopicError) end end end diff --git a/spec/lib/karafka/web/processing/consumers/metrics_spec.rb b/spec/lib/karafka/web/processing/consumers/metrics_spec.rb index 996d299c..c9efa256 100644 --- a/spec/lib/karafka/web/processing/consumers/metrics_spec.rb +++ b/spec/lib/karafka/web/processing/consumers/metrics_spec.rb @@ -15,6 +15,16 @@ it { expect { metrics }.to raise_error(expected_error) } end + context 'when metrics topic does not exist' do + let(:expected_error) do + ::Karafka::Web::Errors::Processing::MissingConsumersMetricsTopicError + end + + before { Karafka::Web.config.topics.consumers.metrics = SecureRandom.uuid } + + it { expect { metrics }.to raise_error(expected_error) } + end + context 'when current state exists' do before { produce(metrics_topic, fixture) } diff --git a/spec/lib/karafka/web/processing/consumers/state_spec.rb b/spec/lib/karafka/web/processing/consumers/state_spec.rb index 49bb9f2e..2a3beb76 100644 --- a/spec/lib/karafka/web/processing/consumers/state_spec.rb +++ b/spec/lib/karafka/web/processing/consumers/state_spec.rb @@ -15,6 +15,16 @@ it { expect { state }.to raise_error(expected_error) } end + context 'when states topic does not exist' do + let(:expected_error) do + ::Karafka::Web::Errors::Processing::MissingConsumersStatesTopicError + end + + before { Karafka::Web.config.topics.consumers.states = SecureRandom.uuid } + + it { expect { state }.to raise_error(expected_error) } + end + context 'when current state exists' do before { produce(states_topic, fixture) }