From a34790085ae7f79615f7225911ba9cf8ceb8f3e2 Mon Sep 17 00:00:00 2001
From: Yuki Morishita <yukim@apache.org>
Date: Mon, 12 Aug 2013 11:07:53 -0500
Subject: [PATCH] suppress sending too many progress notifications

---
 .../management/StreamEventJMXNotifier.java    | 26 +++++++++++++------
 1 file changed, 18 insertions(+), 8 deletions(-)

diff --git a/src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java b/src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java
index f8c54ec022..01ca9a166b 100644
--- a/src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java
+++ b/src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java
@@ -21,17 +21,19 @@ import java.util.concurrent.atomic.AtomicLong;
 import javax.management.Notification;
 import javax.management.NotificationBroadcasterSupport;
 
-import org.apache.cassandra.streaming.StreamEvent;
-import org.apache.cassandra.streaming.StreamEventHandler;
-import org.apache.cassandra.streaming.StreamManagerMBean;
-import org.apache.cassandra.streaming.StreamState;
+import org.apache.cassandra.streaming.*;
 
 /**
  */
 public class StreamEventJMXNotifier extends NotificationBroadcasterSupport implements StreamEventHandler
 {
+    // interval in millisec to use for progress notification
+    private static final long PROGRESS_NOTIFICATION_INTERVAL = 1000;
+
     private final AtomicLong seq = new AtomicLong();
 
+    private long progressLastSent;
+
     public void handleStreamEvent(StreamEvent event)
     {
         Notification notif = null;
@@ -49,10 +51,18 @@ public class StreamEventJMXNotifier extends NotificationBroadcasterSupport imple
                 notif.setUserData(SessionCompleteEventCompositeData.toCompositeData((StreamEvent.SessionCompleteEvent) event));
                 break;
             case FILE_PROGRESS:
-                notif = new Notification(StreamEvent.ProgressEvent.class.getCanonicalName(),
-                                         StreamManagerMBean.OBJECT_NAME,
-                                         seq.getAndIncrement());
-                notif.setUserData(ProgressInfoCompositeData.toCompositeData(event.planId, ((StreamEvent.ProgressEvent) event).progress));
+                ProgressInfo progress = ((StreamEvent.ProgressEvent) event).progress;
+                long current = System.currentTimeMillis();
+                if (current - progressLastSent >= PROGRESS_NOTIFICATION_INTERVAL || progress.isCompleted())
+                {
+                    notif = new Notification(StreamEvent.ProgressEvent.class.getCanonicalName(),
+                                             StreamManagerMBean.OBJECT_NAME,
+                                             seq.getAndIncrement());
+                    notif.setUserData(ProgressInfoCompositeData.toCompositeData(event.planId, progress));
+                    progressLastSent = System.currentTimeMillis();
+                } else {
+                    return;
+                }
                 break;
         }
         sendNotification(notif);