Netflix · metacret · May 20, 2014 · May 19, 2014 · May 19, 2014 · May 20, 2014
diff --git a/build.gradle b/build.gradle
@@ -111,14 +111,14 @@ project(':suro-server') {
     compile project(':suro-core')
     compile project(':suro-client')
     compile project(':suro-kafka')
-
-    compile 'net.java.dev.jets3t:jets3t:0.9.0'
+    compile project(':suro-localfile')
+    compile project(':suro-s3')
 
     compile 'org.eclipse.jetty:jetty-servlet:7.6.7.v20120910'
     compile 'com.sun.jersey.contribs:jersey-guice:1.11'
     compile 'com.sun.jersey:jersey-bundle:1.11'
-    compile 'com.amazonaws:aws-java-sdk:1.4.7'
     compile 'javax.servlet:servlet-api:2.5'
+    compile 'com.amazonaws:aws-java-sdk:1.4.7'
 
     compile 'commons-cli:commons-cli:1.2'
 
@@ -174,3 +174,25 @@ project(':suro-kafka') {
     testCompile project(':suro-client')
   }
 }
+
+project(':suro-s3') {
+    dependencies {
+        compile project(':suro-core')
+        compile project(':suro-localfile')
+
+        compile 'net.java.dev.jets3t:jets3t:0.9.0'
+        compile 'com.amazonaws:aws-java-sdk:1.4.7'
+
+        testCompile project(':suro-client').sourceSets.test.output
+    }
+}
+
+project(':suro-localfile') {
+    dependencies {
+        compile project(':suro-core')
+
+        compile project(':suro-client')
+
+        testCompile project(':suro-client').sourceSets.test.output
+    }
+}
diff --git a/gradle/wrapper/gradle-wrapper.properties b/gradle/wrapper/gradle-wrapper.properties
@@ -1,4 +1,8 @@
+<<<<<<< HEAD
+#Sat May 17 22:50:37 PDT 2014
+=======
 #Mon May 19 18:35:38 PDT 2014
+>>>>>>> upstream/master
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME

diff --git a/settings.gradle b/settings.gradle
@@ -1,2 +1,2 @@
 rootProject.name='suro'
-include 'suro-core', 'suro-client', 'suro-server', 'suro-integration-test', 'suro-kafka'
+include 'suro-core', 'suro-client', 'suro-server', 'suro-integration-test', 'suro-kafka', 'suro-s3', 'suro-localfile'
diff --git a/...rc/main/java/com/netflix/suro/TagKey.java → ...rc/main/java/com/netflix/suro/TagKey.java b/...rc/main/java/com/netflix/suro/TagKey.java → ...rc/main/java/com/netflix/suro/TagKey.java
diff --git a/suro-core/src/main/java/com/netflix/suro/queue/TrafficController.java b/suro-core/src/main/java/com/netflix/suro/queue/TrafficController.java
@@ -0,0 +1,9 @@
+package com.netflix.suro.queue;
+
+public interface TrafficController {
+    void stopTakingTraffic();
+
+    void startTakingTraffic();
+
+    int getStatus();
+}
diff --git a/...netflix/suro/sink/notice/QueueNotice.java → ...netflix/suro/sink/notice/QueueNotice.java b/...netflix/suro/sink/notice/QueueNotice.java → ...netflix/suro/sink/notice/QueueNotice.java
diff --git a/...uro/sink/localfile/FileNameFormatter.java → ...uro/sink/localfile/FileNameFormatter.java b/...uro/sink/localfile/FileNameFormatter.java → ...uro/sink/localfile/FileNameFormatter.java
diff --git a/...tflix/suro/sink/localfile/FileWriter.java → ...tflix/suro/sink/localfile/FileWriter.java b/...tflix/suro/sink/localfile/FileWriter.java → ...tflix/suro/sink/localfile/FileWriter.java
@@ -42,29 +42,29 @@ public interface FileWriter {
      * {@link FileNameFormatter}
      *
      * @param outputDir the directory where the file is located
-     * @throws IOException
+     * @throws java.io.IOException
      */
     void open(String outputDir) throws IOException;
 
     /**
      * @return the file length
-     * @throws IOException
+     * @throws java.io.IOException
      */
     long getLength() throws IOException;
     void writeTo(Message message) throws IOException;
 
     /**
      * Flush all data to the disk
      *
-     * @throws IOException
+     * @throws java.io.IOException
      */
     void sync() throws IOException;
 
     /**
      * Close the current file, create and open the new file.
      *
      * @param newPath The path that points to the newly rotated file.
-     * @throws IOException
+     * @throws java.io.IOException
      */
     void rotate(String newPath) throws IOException;
     FileSystem getFS();
@@ -77,7 +77,7 @@ public interface FileWriter {
      *
      * @param oldName The name of the file when it is not done.
      * @param newName The new name of the file after is is marked as done.
-     * @throws IOException
+     * @throws java.io.IOException
      */
     void setDone(String oldName, String newName) throws IOException;
 }
diff --git a/...x/suro/sink/localfile/FileWriterBase.java → ...x/suro/sink/localfile/FileWriterBase.java b/...x/suro/sink/localfile/FileWriterBase.java → ...x/suro/sink/localfile/FileWriterBase.java
@@ -62,7 +62,7 @@ public FileWriterBase(String codecClass, Logger log, Configuration conf) {
      *
      * @param oldName
      * @param newName
-     * @throws IOException
+     * @throws java.io.IOException
      */
     public void setDone(String oldName, String newName) throws IOException {
         Path oldPath = new Path(oldName);
@@ -104,7 +104,7 @@ public static CompressionCodec createCodecInstance(String codecClass) throws Cla
      *
      * @param newPath
      * @return
-     * @throws IOException
+     * @throws java.io.IOException
      */
     public SequenceFile.Writer createSequenceFile(String newPath) throws IOException {
         if (codec != null) {
@@ -125,7 +125,7 @@ fs, conf, new Path(newPath),
      *
      * @param path
      * @return
-     * @throws IOException
+     * @throws java.io.IOException
      */
     public FSDataOutputStream createFSDataOutputStream(String path) throws IOException {
         return fs.create(new Path(path), false);
@@ -137,7 +137,7 @@ public FSDataOutputStream createFSDataOutputStream(String path) throws IOExcepti
      *
      * @param outputStream
      * @return
-     * @throws IOException
+     * @throws java.io.IOException
      */
     public DataOutputStream createDataOutputStream(FSDataOutputStream outputStream) throws IOException {
         if (codec != null) {

diff --git a/...ix/suro/sink/localfile/LocalFileSink.java → ...ix/suro/sink/localfile/LocalFileSink.java b/...ix/suro/sink/localfile/LocalFileSink.java → ...ix/suro/sink/localfile/LocalFileSink.java
@@ -30,7 +30,7 @@
 import com.netflix.suro.message.MessageContainer;
 import com.netflix.suro.sink.notice.QueueNotice;
 import com.netflix.suro.sink.notice.Notice;
-import com.netflix.suro.queue.MessageSetProcessorManager;
+import com.netflix.suro.queue.TrafficController;
 import com.netflix.suro.sink.QueuedSink;
 import com.netflix.suro.sink.Sink;
 import com.netflix.suro.queue.MemoryQueue4Sink;
@@ -51,8 +51,8 @@
 /**
  * LocalFileSink appends messages to the file in local file system and rotates
  * the file when the file size reaches to the threshold or in the regular basis
- * whenever it comes earlier. When {@link SpaceChecker} checks not enough disk
- * space, it triggers {@link com.netflix.suro.queue.MessageSetProcessorManager} not to take the traffic anymore.
+ * whenever it comes earlier. When {@link com.netflix.suro.sink.localfile.LocalFileSink.SpaceChecker} checks not enough disk
+ * space, it triggers {@link com.netflix.suro.queue.TrafficController} not to take the traffic anymore.
  *
  * @author jbae
  */
@@ -71,7 +71,7 @@ public class LocalFileSink extends QueuedSink implements Sink {
     private final int minPercentFreeDisk;
     private final Notice<String> notice;
 
-    private MessageSetProcessorManager messageSetProcessorManager;
+    private TrafficController trafficController;
     private SpaceChecker spaceChecker;
 
     private String filePath;
@@ -97,7 +97,7 @@ public LocalFileSink(
             @JsonProperty("queue4Sink") MessageQueue4Sink queue4Sink,
             @JsonProperty("batchSize") int batchSize,
             @JsonProperty("batchTimeout") int batchTimeout,
-            @JacksonInject("queueManager") MessageSetProcessorManager messageSetProcessorManager,
+            @JacksonInject("queueManager") TrafficController trafficController,
             @JacksonInject("spaceChecker") SpaceChecker spaceChecker) {
         if (!outputDir.endsWith("/")) {
             outputDir += "/";
@@ -110,7 +110,7 @@ public LocalFileSink(
         this.rotationPeriod = new Period(rotationPeriod == null ? "PT2m" : rotationPeriod);
         this.minPercentFreeDisk = minPercentFreeDisk == 0 ? 85 : minPercentFreeDisk;
         this.notice = notice == null ? new QueueNotice<String>() : notice;
-        this.messageSetProcessorManager = messageSetProcessorManager;
+        this.trafficController = trafficController;
         this.spaceChecker = spaceChecker;
 
         Monitors.registerObject(LocalFileSink.class.getSimpleName() + "-" + outputDir.replace('/', '_'), this);
@@ -127,8 +127,23 @@ public void open() {
             if (spaceChecker == null) {
                 spaceChecker = new SpaceChecker(minPercentFreeDisk, outputDir);
             }
-            if (messageSetProcessorManager == null) {
-                messageSetProcessorManager = new MessageSetProcessorManager();
+            if (trafficController == null) {
+                trafficController = new TrafficController() {
+                    @Override
+                    public void stopTakingTraffic() {
+
+                    }
+
+                    @Override
+                    public void startTakingTraffic() {
+
+                    }
+
+                    @Override
+                    public int getStatus() {
+                        return 0;
+                    }
+                };
             }
 
             notice.init();
@@ -191,17 +206,17 @@ private void rotate() throws IOException {
         nextRotation = new DateTime().plus(rotationPeriod).getMillis();
 
         if (!spaceChecker.hasEnoughSpace()) {
-            messageSetProcessorManager.stopTakingTraffic();
+            trafficController.stopTakingTraffic();
         } else {
-            messageSetProcessorManager.startTakingTraffic();
+            trafficController.startTakingTraffic();
         }
     }
 
     /**
      * Before polling messages from the queue, it should check whether to rotate
      * the file and start to write to new file.
      *
-     * @throws IOException
+     * @throws java.io.IOException
      */
     @Override
     protected void beforePolling() throws IOException {
@@ -218,7 +233,7 @@ protected void beforePolling() throws IOException {
      * commit the queue and clear messages
      *
      * @param msgList
-     * @throws IOException
+     * @throws java.io.IOException
      */
     @Override
     protected void write(List<Message> msgList) throws IOException {

diff --git a/.../suro/sink/localfile/MessageWritable.java → .../suro/sink/localfile/MessageWritable.java b/.../suro/sink/localfile/MessageWritable.java → .../suro/sink/localfile/MessageWritable.java
diff --git a/...ro/sink/localfile/SequenceFileViewer.java → ...ro/sink/localfile/SequenceFileViewer.java b/...ro/sink/localfile/SequenceFileViewer.java → ...ro/sink/localfile/SequenceFileViewer.java
diff --git a/...ro/sink/localfile/SequenceFileWriter.java → ...ro/sink/localfile/SequenceFileWriter.java b/...ro/sink/localfile/SequenceFileWriter.java → ...ro/sink/localfile/SequenceFileWriter.java
diff --git a/...x/suro/sink/localfile/TextFileWriter.java → ...x/suro/sink/localfile/TextFileWriter.java b/...x/suro/sink/localfile/TextFileWriter.java → ...x/suro/sink/localfile/TextFileWriter.java
diff --git a/suro-localfile/src/test/java/com/netflix/suro/sink/localfile/SuroSinkPlugin.java b/suro-localfile/src/test/java/com/netflix/suro/sink/localfile/SuroSinkPlugin.java
@@ -0,0 +1,18 @@
+package com.netflix.suro.sink.localfile;
+
+import com.netflix.suro.SuroPlugin;
+import com.netflix.suro.sink.SuroSink;
+import com.netflix.suro.sink.notice.NoNotice;
+import com.netflix.suro.sink.notice.QueueNotice;
+
+public class SuroSinkPlugin extends SuroPlugin {
+    @Override
+    protected void configure() {
+        this.addSinkType(LocalFileSink.TYPE, LocalFileSink.class);
+
+        this.addSinkType(SuroSink.TYPE, SuroSink.class);
+
+        this.addNoticeType(NoNotice.TYPE, NoNotice.class);
+        this.addNoticeType(QueueNotice.TYPE, QueueNotice.class);
+    }
+}
diff --git a/...sink/localfile/TestFileNameFormatter.java → ...sink/localfile/TestFileNameFormatter.java b/...sink/localfile/TestFileNameFormatter.java → ...sink/localfile/TestFileNameFormatter.java