Fixed issue where null values cause exception, and make sure state save exceptions bubble up

Author: s-nel

src/main/java/storm/trident/state/postgresql/PostgresqlState.java

@@ -27,6 +27,7 @@
 import storm.trident.state.map.OpaqueMap;
 import storm.trident.state.map.TransactionalMap;
 import backtype.storm.task.IMetricsContext;
+import backtype.storm.topology.FailedException;
 
 import com.google.common.base.Function;
 import com.google.common.base.Joiner;
@@ -136,7 +137,7 @@ public void multiPut(final List<List<Object>> keys, final List<T> values) {
           .append(buildColumns())
           .append(") AS (")
             .append("VALUES ")
-            .append(Joiner.on(", ").join(repeat("(" + Joiner.on(",").join(repeat("?", paramCount)) + ")", pkeys.size())))
+            .append(Joiner.on(", ").join(repeat("(" + buildValueParams() + ")", pkeys.size())))
           .append("),")
           .append("updated AS (")
             .append("UPDATE ").append(config.getTable()).append(" t ")
@@ -198,6 +199,7 @@ public String apply(final String col) {
         ps.execute();
       } catch (final SQLException ex) {
         logger.error("Multiput update failed", ex);
+        throw new FailedException(ex);
       } finally {
         if (ps != null) {
           try {
@@ -217,6 +219,16 @@ private String buildColumns() {
     return Joiner.on(",").join(cols);
   }
 
+  private String buildValueParams() {
+    final List<String> types = Lists.newArrayList(config.getKeyTypes()); // the columns for the composite unique key
+    types.addAll(getValueTypes());
+    List<String> withCast = new ArrayList<>();
+    for(String type : types) {
+      withCast.add(String.format("?::%s", type));
+    }
+    return Joiner.on(",").join(withCast);
+  }
+
   private String buildKeyQuery(final int n) {
     final String single = "(" + Joiner.on(" AND ").join(Lists.transform(Arrays.asList(config.getKeyColumns()), new Function<String, String>() {
       @Override
@@ -243,6 +255,17 @@ public String apply(final String field) {
     return cols;
   }
 
+  private List<String> getValueTypes() {
+    final List<String> cols = Lists.newArrayList(config.getValueTypes()); // the columns storing the values
+    if (StateType.OPAQUE.equals(config.getType()) || StateType.TRANSACTIONAL.equals(config.getType())) {
+      cols.add("bigint");
+    }
+    if (StateType.OPAQUE.equals(config.getType())) {
+      cols.addAll(Lists.newArrayList(config.getValueTypes()));
+    }
+    return cols;
+  }
+
   /**
    * run the multi get query, passing in the list of keys and returning key tuples mapped to value tuples
    *

src/main/java/storm/trident/state/postgresql/PostgresqlStateConfig.java

@@ -12,6 +12,8 @@ public class PostgresqlStateConfig implements Serializable {
 	private StateType type = StateType.OPAQUE;
 	private String[] keyColumns;
 	private String[] valueColumns;
+	private String[] valueTypes;
+	private String[] keyTypes;
 	private int batchSize = DEFAULT_BATCH_SIZE;
 	private int cacheSize = DEFAULT_CACHE_SIZE;
 
@@ -74,4 +76,19 @@ public void setBatchSize(int batchSize) {
 		this.batchSize = batchSize;
 	}
 
+	public String[] getValueTypes() {
+	    return valueTypes;
+	}
+
+	public void setValueTypes(String[] valueTypes) {
+	    this.valueTypes = valueTypes;
+	}
+
+	public String[] getKeyTypes() {
+        return keyTypes;
+    }
+
+    public void setKeyTypes(String[] keyTypes) {
+        this.keyTypes = keyTypes;
+    }
 }