alert when resetting offsets (apache#3931)

* alert when resetting offsets * add more data to alerts
Gamehu · Feb 13, 2017 · 1f263fe · 1f263fe
1 parent c1eee9b
commit 1f263fe
Showing 1 changed file with 6 additions and 3 deletions.
diff --git a/...ons-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java b/...ons-core/kafka-indexing-service/src/main/java/io/druid/indexing/kafka/KafkaIndexTask.java
@@ -1039,10 +1039,11 @@ private void possiblyResetOffsetsOrWait(
     }
   }
 
-  private void sendResetRequestAndWait(Map<TopicPartition, Long> outOfRangePartitions, TaskToolbox taskToolbox) throws IOException
+  private void sendResetRequestAndWait(Map<TopicPartition, Long> outOfRangePartitions, TaskToolbox taskToolbox)
+      throws IOException
   {
     Map<Integer, Long> partitionOffsetMap = Maps.newHashMap();
-    for (Map.Entry<TopicPartition, Long>  outOfRangePartition: outOfRangePartitions.entrySet()) {
+    for (Map.Entry<TopicPartition, Long> outOfRangePartition : outOfRangePartitions.entrySet()) {
       partitionOffsetMap.put(outOfRangePartition.getKey().partition(), outOfRangePartition.getValue());
     }
     boolean result = taskToolbox.getTaskActionClient()
@@ -1056,7 +1057,9 @@ private void sendResetRequestAndWait(Map<TopicPartition, Long> outOfRangePartiti
                                 ));
 
     if (result) {
-      log.warn("Successfully sent the reset request for partitions [%s], waiting to be killed", partitionOffsetMap.keySet());
+      log.makeAlert("Resetting Kafka offsets for datasource [%s]", getDataSource())
+         .addData("partitions", partitionOffsetMap.keySet())
+         .emit();
       // wait for being killed by supervisor
       try {
         Thread.sleep(Long.MAX_VALUE);