Extended Configuration for binary data.

Fixed de/encoding problem in TaskConfig.

Author: sewen

nephele/nephele-common/src/main/java/eu/stratosphere/nephele/configuration/Configuration.java

@@ -24,6 +24,7 @@
 import java.util.Map;
 import java.util.Set;
 
+import org.apache.commons.codec.binary.Base64;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 
@@ -349,6 +350,48 @@ public void setFloat(final String key, final float value) {
 			this.confData.put(key, Float.toString(value));
 		}
 	}
+	
+
+	/**
+	 * Returns the value associated with the given key as a byte array.
+	 * 
+	 * @param key
+	 *        The key pointing to the associated value.
+	 * @param defaultValue
+	 *        The default value which is returned in case there is no value associated with the given key.
+	 * @return the (default) value associated with the given key.
+	 */
+	public byte[] getBytes(final String key, final byte[] defaultValue) {
+		final String encoded;
+		synchronized (this.confData) {
+			encoded = this.confData.get(key);
+		}
+		if (encoded == null) {
+			return defaultValue;
+		}
+		
+		return Base64.decodeBase64(encoded.getBytes());
+	}
+	
+	/**
+	 * Adds the given byte array to the configuration object. If key is <code>null</code> then nothing is added.
+	 * 
+	 * @param key
+	 *        The key under which the bytes are added.
+	 * @param bytes
+	 *        The bytes to be added.
+	 */
+	public void setBytes(final String key, final byte[] bytes) {
+		if (key == null) {
+			LOG.warn("Cannot set boolean: Given key is null!");
+			return;
+		}
+		
+		final String encoded = new String(Base64.encodeBase64(bytes));
+		synchronized (this.confData) {
+			this.confData.put(key, encoded);
+		}
+	}
 
 	/**
 	 * Returns the keys of all key/value pairs stored inside this

pact/pact-runtime/src/main/java/eu/stratosphere/pact/runtime/task/util/TaskConfig.java

@@ -434,8 +434,8 @@ public void setOutputDataDistribution(DataDistribution distribution)
 		} catch (IOException e) {
 			throw new RuntimeException("Error serializing the DataDistribution: " + e.getMessage(), e);
 		}
-		final String stateEncoded = baos.toString();
-		this.config.setString(OUTPUT_DATA_DISTRIBUTION_STATE, stateEncoded);
+
+		this.config.setBytes(OUTPUT_DATA_DISTRIBUTION_STATE, baos.toByteArray());
 	}
 
 	public DataDistribution getOutputDataDistribution(final ClassLoader cl) throws ClassNotFoundException
@@ -455,13 +455,13 @@ public DataDistribution getOutputDataDistribution(final ClassLoader cl) throws C
 
 		final DataDistribution distribution = InstantiationUtil.instantiate(clazz, DataDistribution.class);
 
-		final String stateEncoded = this.config.getString(OUTPUT_DATA_DISTRIBUTION_STATE, null);
+		final byte[] stateEncoded = this.config.getBytes(OUTPUT_DATA_DISTRIBUTION_STATE, null);
 		if (stateEncoded == null) {
 			throw new CorruptConfigurationException(
 						"The configuration contained the data distribution type, but no serialized state.");
 		}
 
-		final ByteArrayInputStream bais = new ByteArrayInputStream(stateEncoded.getBytes());
+		final ByteArrayInputStream bais = new ByteArrayInputStream(stateEncoded);
 		final DataInputStream in = new DataInputStream(bais);
 
 		try {