Allow offset editing and per partition pausing/resuming

CHANGELOG.md

@@ -1,10 +1,21 @@
 # Karafka Web Changelog
 
 ## 0.11.0 (Unreleased)
+- [Feature] Provide ability to pause/resume partitions on running consumers via the UI (Pro).
+- [Feature] Provide ability to edit offsets of running consumers (Pro).
 - [Feature] Support consumers that have mismatching schema.
 - [Feature] Provide ability to navigate to a timestamp in the Explorer (Pro).
+- [Enhancement] Improve handling of post-submit redirects.
 - [Enhancement] Provide better support for fully transactional consumers.
 - [Enhancement] Error out when `#setup` is called after `#enable!`.
+- [Change] Remove per-consumer process duplicated details from Subscriptions and Jobs tabs.
+- [Fix] Fix incorrect names in some of the tables headers.
+- [Fix] Normalize position of commanding buttons in regards to other UI elements.
+- [Fix] Fix incorrect indentation of some of the info messages.
+- [Fix] Fix tables headers inconsistent alignments.
+- [Fix] Fix incorrect warning box header color in the dark mode.
+- [Fix] Fix missing breadcrumbs on the consumers overview page.
+- [Fix] Fix a case where disabled buttons would be enabled back too early.
 - [Fix] The recent page breadcrumbs and offset id are not refreshed on change.
 - [Fix] Direct URL access with too big partition causes librdkafka crash.
 - [Fix] Fix incorrect breadcrumbs for pending consumer jobs.

lib/karafka/web/pro/commanding/commands/base.rb

@@ -11,20 +11,38 @@ module Commanding
         module Commands
           # Base for all the commands
           class Base
+            class << self
+              attr_accessor :name
+            end
+
             # @return [Hash]
-            attr_reader :params
+            attr_reader :command
 
-            # @param params [Hash] command details (if any). Some commands may require extra
+            # @param command [Command] command details (if any). Some commands may require extra
             #   details to work. They can be obtained from here.
-            def initialize(params)
-              @params = params
+            def initialize(command)
+              @command = command
+            end
+
+            # Executes the command after receiving it.
+            def call
+              raise NotImlementedError, 'Please implement in a subclass'
             end
 
             private
 
-            # @return [String] current process id
-            def process_id
-              @process_id ||= ::Karafka::Web.config.tracking.consumers.sampler.process_id
+            # Dispatches the acceptance message back to Kafka as a confirmation
+            #
+            # @param params [Hash] hash with the acceptance message details
+            def acceptance(params)
+              Dispatcher.acceptance(self.class.name, process_id, params)
+            end
+
+            # Dispatches the result message back to Kafka with execution details
+            #
+            # @param params [Hash] hash with the result message details
+            def result(params)
+              Dispatcher.result(self.class.name, process_id, params)
             end
 
             # @return [Boolean] Is given process to which a command was sent operating in an
@@ -34,6 +52,11 @@ def process_id
             def standalone?
               Karafka::Server.execution_mode == :standalone
             end
+
+            # @return [String] id of the current consumer process
+            def process_id
+              ::Karafka::Web.config.tracking.consumers.sampler.process_id
+            end
           end
         end
       end

lib/karafka/web/pro/commanding/commands/consumers/quiet.rb

@@ -0,0 +1,33 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        # Namespace for commands the process can react to
+        module Commands
+          # Namespace for commands related to consumers themselves
+          module Consumers
+            # Sends a signal to quiet the consumer
+            # @note Does not work in an embedded mode because we do not own the Ruby process.
+            class Quiet < Base
+              self.name = 'consumers.quiet'
+
+              # Performs the command if not in embedded mode
+              def call
+                return unless standalone?
+
+                ::Process.kill('TSTP', ::Process.pid)
+
+                result(status: 'applied')
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/commands/consumers/stop.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        # Namespace for commands the process can react to
+        module Commands
+          module Consumers
+            # Sends a signal to stop the process
+            # @note Does not work in an embedded mode because we do not own the Ruby process.
+            class Stop < Base
+              self.name = 'consumers.stop'
+
+              # Performs the command if not in embedded mode
+              def call
+                return unless standalone?
+
+                ::Process.kill('QUIT', ::Process.pid)
+
+                result(status: 'applied')
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/commands/consumers/trace.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        # Namespace for commands the process can react to
+        module Commands
+          module Consumers
+            # Collects all backtraces from the available Ruby threads and publishes their details
+            #   back to Kafka for debug.
+            class Trace < Base
+              self.name = 'consumers.trace'
+
+              # Runs tracing and publishes result back to Kafka
+              def call
+                threads = {}
+
+                Thread.list.each do |thread|
+                  tid = (thread.object_id ^ ::Process.pid).to_s(36)
+                  t_d = threads[tid] = {}
+                  t_d[:label] = "Thread TID-#{tid} #{thread.name}"
+                  t_d[:backtrace] = (thread.backtrace || ['<no backtrace available>']).join("\n")
+                end
+
+                result(threads)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/commands/partitions/pause.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        module Commands
+          # Namespace for post-fetch command execution
+          module Partitions
+            # Delegates the pause request into the partition changes tracker and dispatches the
+            # acceptance message back to Kafka
+            class Pause < Base
+              self.name = 'partitions.pause'
+
+              # Delegates the pause request to async handling
+              def call
+                Handlers::Partitions::Tracker.instance << command
+
+                acceptance(command.to_h)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/commands/partitions/resume.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        module Commands
+          module Partitions
+            # Delegates the resuming of the processing
+            class Resume < Base
+              self.name = 'partitions.resume'
+
+              # Dispatches the seek request into the appropriate filter and indicates that the
+              # seeking is in an acceptance state
+              def call
+                Handlers::Partitions::Tracker.instance << command
+
+                # Publish back info on who did this with all the details for inspection
+                acceptance(command.to_h)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/commands/partitions/seek.rb

@@ -0,0 +1,30 @@
+# frozen_string_literal: true
+
+# This code is part of Karafka Pro, a commercial component not licensed under LGPL.
+# See LICENSE for details.
+
+module Karafka
+  module Web
+    module Pro
+      module Commanding
+        module Commands
+          module Partitions
+            # Topic partition seek command request handler
+            class Seek < Base
+              self.name = 'partitions.seek'
+
+              # Dispatches the seek request into the appropriate filter and indicates that the
+              # seeking is in an acceptance state
+              def call
+                Handlers::Partitions::Tracker.instance << command
+
+                # Publish back info on who did this with all the details for inspection
+                acceptance(command.to_h)
+              end
+            end
+          end
+        end
+      end
+    end
+  end
+end

lib/karafka/web/pro/commanding/config.rb

@@ -35,6 +35,10 @@ class Config
             'auto.offset.reset': 'latest'
           }
 
+          setting :listeners, default: [
+            Handlers::Partitions::Listener.new
+          ]
+
           configure
         end
       end