stop and requeue message migration job periodically

break it up into 5 minute chunks

refs #CORE-2558

Change-Id: I00b6d7ffb548a1543e56e6397aeff177a8724f76
Reviewed-on: https://gerrit.instructure.com/185084
Reviewed-by: Cody Cutrer <cody@instructure.com>
QA-Review: Cody Cutrer <cody@instructure.com>
Product-Review: Cody Cutrer <cody@instructure.com>
Tested-by: Jenkins

gems/canvas_partman/lib/canvas_partman/partition_manager/by_date.rb

@@ -33,8 +33,11 @@ module CanvasPartman
         ensure_partitions(advance_partitions)
       end
 
-      def migrate_data_to_partitions(batch_size: 1000)
+      def migrate_data_to_partitions(batch_size: 1000, timeout: nil)
+        start_time = Time.now
         loop do
+          return false if timeout && (start_time + timeout) < Time.now
+
           id_dates = base_class.from("ONLY #{base_class.quoted_table_name}").
             order(base_class.partitioning_field).
             limit(batch_size).
@@ -51,6 +54,7 @@ module CanvasPartman
             SQL
           end
         end
+        true
       end
 
       def ensure_partitions(advance = 1)

lib/data_fixup/migrate_messages_to_partitions.rb

@@ -27,6 +27,14 @@ module DataFixup::MigrateMessagesToPartitions
 
     partman = CanvasPartman::PartitionManager.create(Message)
 
-    partman.migrate_data_to_partitions
+    unless partman.migrate_data_to_partitions(timeout: 5.minutes)
+      self.requeue # timed out
+    end
+  end
+
+  def self.requeue
+    self.send_later_if_production_enqueue_args(:run,
+      priority: Delayed::LOWER_PRIORITY,
+      strand: "partition_messages:#{Shard.current.database_server.id}")
   end
 end