Fixing the issue with Default Partitioner

The code in the 08 branch was forcing the use of the Default partitioner. Syncing this with the 07 branch.
hershs · Aug 12, 2013 · 9ab26ab · 9ab26ab
1 parent 2b94ef4
commit 9ab26ab
Showing 1 changed file with 10 additions and 2 deletions.
diff --git a/camus-etl-kafka/src/main/java/com/linkedin/camus/etl/kafka/mapred/EtlMultiOutputFormat.java b/camus-etl-kafka/src/main/java/com/linkedin/camus/etl/kafka/mapred/EtlMultiOutputFormat.java
@@ -200,15 +200,23 @@ public String getWorkingFileName(JobContext context, EtlKey key) throws IOExcept
         return "data." + key.getTopic().replaceAll("\\.", "_") + "." + key.getLeaderId() + "." + key.getPartition() + "." + partitioner.encodePartition(context, key);
     }
 
-    public static Partitioner getDefaultPartitioner(JobContext job) {
+/*    public static Partitioner getDefaultPartitioner(JobContext job) {
         if(partitionersByTopic.get(ETL_DEFAULT_PARTITIONER_CLASS) == null) {
             //List<Partitioner> partitioners = job.getConfiguration().getInstances(ETL_DEFAULT_PARTITIONER_CLASS, com.linkedin.camus.coders.Partitioner.class);
             List<Partitioner> partitioners = new ArrayList<Partitioner>();
             partitioners.add(new DefaultPartitioner());
             partitionersByTopic.put(ETL_DEFAULT_PARTITIONER_CLASS, partitioners.get(0));
         }
         return partitionersByTopic.get(ETL_DEFAULT_PARTITIONER_CLASS);
-    }
+    }*/
+
+    public static Partitioner getDefaultPartitioner(JobContext job) {
+        if(partitionersByTopic.get(ETL_DEFAULT_PARTITIONER_CLASS) == null) {
+            List<Partitioner> partitioners = job.getConfiguration().getInstances(ETL_DEFAULT_PARTITIONER_CLASS, com.linkedin.camus.coders.Partitioner.class);
+            partitionersByTopic.put(ETL_DEFAULT_PARTITIONER_CLASS, partitioners.get(0));
+        }
+        return partitionersByTopic.get(ETL_DEFAULT_PARTITIONER_CLASS);
+    }    
 
     public static Partitioner getPartitioner(JobContext job, String topicName) throws IOException {
         String customPartitionerProperty = ETL_DEFAULT_PARTITIONER_CLASS + "." + topicName;