Merge pull request Netflix#178 from Netflix/ISSUE-176

control SuroInput pause with just one time, not accumulative one
katoomba-demo · Jan 15, 2015 · c6db346 · c6db346
2 parents 8a3e7c3 + 2614897
commit c6db346
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 5 deletions.
diff --git a/suro-kafka-consumer/src/main/java/com/netflix/suro/input/kafka/KafkaConsumer.java b/suro-kafka-consumer/src/main/java/com/netflix/suro/input/kafka/KafkaConsumer.java
@@ -71,6 +71,8 @@ public String getId() {
 
     private AtomicLong pausedTime = new AtomicLong(0);
 
+    public static long MAX_PAUSE = 1000; // not final for the test
+
     @Override
     public void start() throws Exception {
         executor = Executors.newCachedThreadPool(
@@ -93,10 +95,10 @@ public void start() throws Exception {
                         public void run() {
                             while (running) {
                                 try {
-                                    long pause = pausedTime.get();
+                                    long pause = Math.min(pausedTime.get(), MAX_PAUSE);
                                     if (pause > 0) {
                                         Thread.sleep(pause);
-                                        pausedTime.addAndGet(-pause);
+                                        pausedTime.set(0);
                                     }
                                     byte[] message = iterator.next().message();
                                     router.process(

diff --git a/suro-kafka-consumer/src/test/java/com/netflix/suro/input/kafka/TestKafkaConsumer.java b/suro-kafka-consumer/src/test/java/com/netflix/suro/input/kafka/TestKafkaConsumer.java
@@ -76,6 +76,7 @@ public Object answer(InvocationOnMock invocation) throws Throwable {
 
         properties.setProperty("consumer.timeout.ms", "1000");
         KafkaConsumer consumer = new KafkaConsumer(properties, TOPIC_NAME, numPartitions, router, jsonMapper);
+        KafkaConsumer.MAX_PAUSE = 10000; // for testing
 
         consumer.start();
         latch.await(1000 * 5, TimeUnit.MILLISECONDS);

diff --git a/suro-s3/src/main/java/com/netflix/suro/input/remotefile/S3Consumer.java b/suro-s3/src/main/java/com/netflix/suro/input/remotefile/S3Consumer.java
@@ -93,6 +93,8 @@ public String getId() {
         return id;
     }
 
+    private static final long MAX_PAUSE = 10000;
+
     @Override
     public void start() throws Exception {
         if (s3Service == null) {
@@ -126,7 +128,7 @@ public boolean offer(Runnable runnable) {
                         return true;
                     }
                 },
-                new ThreadFactoryBuilder().setDaemon(true).setNameFormat("S3Consumer-%d").build());
+                new ThreadFactoryBuilder().setDaemon(true).setNameFormat("S3Consumer-" + id + "-%d").build());
 
         notice.init();
 
@@ -136,10 +138,10 @@ public boolean offer(Runnable runnable) {
             public void run() {
                 while (running) {
                     try {
-                        long pause = pausedTime.get();
+                        long pause = Math.min(pausedTime.get(), MAX_PAUSE);
                         if (pause > 0) {
                             Thread.sleep(pause);
-                            pausedTime.addAndGet(-pause);
+                            pausedTime.set(0);
                         }
                         Pair<String, String> msg = notice.peek();
                         if (msg != null) {