natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

suppress sending too many progress notifications

This commit is contained in:
Yuki Morishita 2013-08-12 11:07:53 -05:00
parent a0d6ed1290
commit a34790085a
1 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
src/java/org/apache/cassandra/streaming/management/StreamEventJMXNotifier.java
View File

@ -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);