Fix s3 base path and partition operation (#74)

* Fix s3 base path and partition operation * Fix testGetEvaluatedPartitionsStatic
Nextdoor · May 23, 2018 · 73d4331 · 73d4331
1 parent e0c9646
commit 73d4331
Show file tree

Hide file tree

Showing 5 changed files with 9 additions and 12 deletions.
diff --git a/core/src/main/java/com/nextdoor/bender/partition/PartitionOperation.java b/core/src/main/java/com/nextdoor/bender/partition/PartitionOperation.java
@@ -59,11 +59,7 @@ protected LinkedHashMap<String, String> getPartitions(DeserializedEvent devent)
         }
       }
 
-      if (key != null) {
-        partitions.put(spec.getName(), spec.interpret(key));
-      } else {
-        throw new OperationException("unable to find value for partition " + spec.getName());
-      }
+      partitions.put(spec.getName(), spec.interpret(key));
     }
 
     return partitions;

diff --git a/core/src/main/java/com/nextdoor/bender/partition/PartitionSpec.java b/core/src/main/java/com/nextdoor/bender/partition/PartitionSpec.java
@@ -29,6 +29,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaDefault;
 import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaDescription;
+import com.nextdoor.bender.operation.OperationException;
 import com.nextdoor.bender.utils.Time;
 
 public class PartitionSpec {
@@ -151,7 +152,7 @@ public String interpret(String input) {
 
   protected String getFormattedString(String input) {
     if (input == null) {
-      return input;
+      throw new OperationException("unable to find value for partition '" + this.getName() +"'");
     }
 
     switch (this.stringFormat) {

diff --git a/core/src/test/java/com/nextdoor/bender/partition/PartitionOperationTest.java b/core/src/test/java/com/nextdoor/bender/partition/PartitionOperationTest.java
@@ -23,6 +23,7 @@
 
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.stream.Stream;
@@ -63,8 +64,7 @@ public void testGetEvaluatedPartitionsString() throws FieldNotFoundException {
   @Test
   public void testGetEvaluatedPartitionsStatic() {
     List<PartitionSpec> partitionSpecs = new ArrayList<PartitionSpec>(1);
-    List<String> sources = Arrays.asList("foo");
-    PartitionSpec spec = new PartitionSpec("foo", sources, PartitionSpec.Interpreter.STATIC, "123", 0);
+    PartitionSpec spec = new PartitionSpec("foo", Collections.emptyList(), PartitionSpec.Interpreter.STATIC, "123", 0);
     partitionSpecs.add(spec);
 
     PartitionOperation op = new PartitionOperation(partitionSpecs);
@@ -119,7 +119,7 @@ public void testGetEvaluatedPartitionsStringMultipleFieldsNull() throws FieldNot
     try {
       op.perform(ievent);
     } catch (OperationException e) {
-      assertEquals("unable to find value for partition foo", e.getMessage());
+      assertEquals("unable to find value for partition 'foo'", e.getMessage());
       throw e;
     }
   }
@@ -141,7 +141,7 @@ public void testGetEvaluatedPartitionsFieldNotFoundException() throws FieldNotFo
     try {
       op.perform(ievent);
     } catch (OperationException e) {
-      assertEquals("unable to find value for partition foo", e.getMessage());
+      assertEquals("unable to find value for partition 'foo'", e.getMessage());
       throw e;
     }
   }

diff --git a/transporters/src/main/java/com/nextdoor/bender/ipc/s3/S3Transport.java b/transporters/src/main/java/com/nextdoor/bender/ipc/s3/S3Transport.java
@@ -153,7 +153,7 @@ public void sendBatch(TransportBuffer buffer, LinkedHashMap<String, String> part
 
     if (this.basePath.endsWith("/")) {
       key = this.basePath + key;
-    } else {
+    } else if (!this.basePath.equals("")) {
       key = this.basePath + '/' + key;
     }
 

diff --git a/transporters/src/main/java/com/nextdoor/bender/ipc/s3/S3TransportConfig.java b/transporters/src/main/java/com/nextdoor/bender/ipc/s3/S3TransportConfig.java
@@ -38,7 +38,7 @@ public class S3TransportConfig extends RegionalTransportConfig {
 
   @JsonSchemaDescription("Path to append to S3 keys.")
   @JsonProperty(required = false)
-  private String basePath;
+  private String basePath = "";
 
   @JsonSchemaDescription("Compress files with bz2 compression.")
   @JsonSchemaDefault("true")