Class: Karafka::Web::Pro::Ui::Controllers::Consumers::Partitions::PausesController

Inherits:
BaseController show all
Defined in:
lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb

Overview

Controller for managing partition pauses at the consumer group level.

Instance Method Summary collapse

Methods inherited from ConsumersController

#details, #index, #performance, #subscriptions

Instance Method Details

#create(consumer_group_id, topic, partition_id) ⇒ Object

Dispatches the partition pause command to all processes

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 58

def create(consumer_group_id, topic, partition_id)
  new(consumer_group_id, topic, partition_id)

  # Broadcast to all processes with matchers to filter by consumer group,
  # topic, and partition
  Commanding::Dispatcher.request(
    Commanding::Commands::Partitions::Pause.name,
    {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id,
      # User provides this in seconds, we operate on ms in the system
      duration: params.int(:duration) * 1_000,
      prevent_override: params.bool(:prevent_override)
    },
    matchers: {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id
    }
  )

  redirect(
    :previous,
    success: format_flash(
      "Initiated pause for partition ?#? in consumer group ?",
      topic,
      partition_id,
      consumer_group_id
    )
  )
end

#delete(consumer_group_id, topic, partition_id) ⇒ Object

Dispatches the partition resume command to all processes

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 107

def delete(consumer_group_id, topic, partition_id)
  new(consumer_group_id, topic, partition_id)

  # Broadcast to all processes with matchers to filter by consumer group,
  # topic, and partition
  Commanding::Dispatcher.request(
    Commanding::Commands::Partitions::Resume.name,
    {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id,
      reset_attempts: params.bool(:reset_attempts)
    },
    matchers: {
      consumer_group_id: consumer_group_id,
      topic: topic,
      partition_id: partition_id
    }
  )

  redirect(
    :previous,
    success: format_flash(
      "Initiated resume for partition ?#? in consumer group ?",
      topic,
      partition_id,
      consumer_group_id
    )
  )
end

#edit(consumer_group_id, topic, partition_id) ⇒ Object

Displays the resume configuration form for a partition

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


96
97
98
99
100
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 96

def edit(consumer_group_id, topic, partition_id)
  new(consumer_group_id, topic, partition_id)

  render
end

#new(consumer_group_id, topic, partition_id) ⇒ Object

Displays the pause configuration form for a partition

Parameters:

  • consumer_group_id (String)
  • topic (String)
  • partition_id (Integer) 


47
48
49
50
51
 
# File 'lib/karafka/web/pro/ui/controllers/consumers/partitions/pauses_controller.rb', line 47

def new(consumer_group_id, topic, partition_id)
  bootstrap!(consumer_group_id, topic, partition_id)

  render
end