Minor cleanup and corrections in API classes.

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

@@ -90,7 +90,7 @@ public class Configuration implements IOReadableWritable {
 	 * @see #setClass(String, Class)
 	 */
 	@SuppressWarnings("unchecked")
-	public <T> Class<T> getClass(String key, Class<? extends T> defaultValue, Class<T> ancestor) {
+	public <T> Class<T> getClass(String key, Class<? extends T> defaultValue, Class<? super T> ancestor) {
 		String className = getStringInternal(key);
 		if (className == null) {
 			return (Class<T>) defaultValue;

pact/pact-common/src/main/java/eu/stratosphere/pact/common/stubs/CoGroupStub.java

@@ -48,7 +48,7 @@ public abstract class CoGroupStub extends AbstractStub implements GenericCoGroup
 	 *                   decide whether to retry the task execution.
 	 */
 	@Override
-	public abstract void coGroup(Iterator<PactRecord> records1, Iterator<PactRecord> records2, Collector<PactRecord> out);
+	public abstract void coGroup(Iterator<PactRecord> records1, Iterator<PactRecord> records2, Collector<PactRecord> out) throws Exception;
 	
 	/**
 	 * This method must be overridden by CoGoup UDFs that want to make use of the combining feature
@@ -64,7 +64,7 @@ public abstract class CoGroupStub extends AbstractStub implements GenericCoGroup
 	 *                   decide whether to retry the combiner execution.
 	 */
 	@Override
-	public void combineFirst(Iterator<PactRecord> records, Collector<PactRecord> out) {
+	public void combineFirst(Iterator<PactRecord> records, Collector<PactRecord> out) throws Exception {
 		throw new UnsupportedOperationException();
 	}
 	
@@ -82,7 +82,7 @@ public abstract class CoGroupStub extends AbstractStub implements GenericCoGroup
 	 *                   decide whether to retry the combiner execution.
 	 */
 	@Override
-	public void combineSecond(Iterator<PactRecord> records, Collector<PactRecord> out) {
+	public void combineSecond(Iterator<PactRecord> records, Collector<PactRecord> out) throws Exception {
 		throw new UnsupportedOperationException();
 	}
 }

pact/pact-common/src/main/java/eu/stratosphere/pact/common/stubs/CrossStub.java

@@ -44,5 +44,5 @@ public abstract class CrossStub extends AbstractStub implements GenericCrosser<P
 	 *                   decide whether to retry the task execution.
 	 */
 	@Override
-	public abstract void cross(PactRecord record1, PactRecord record2, Collector<PactRecord> out);
+	public abstract void cross(PactRecord record1, PactRecord record2, Collector<PactRecord> out) throws Exception;
 }

pact/pact-common/src/main/java/eu/stratosphere/pact/common/stubs/MapStub.java

@@ -25,8 +25,6 @@ import eu.stratosphere.pact.generic.stub.GenericMapper;
  * For details on the Map PACT read the documentation of the PACT programming model.
  * <p>
  * For a mapper implementation, the <code>map()</code> method must be implemented.
- * 
- * @author Fabian Hueske
  */
 public abstract class MapStub extends AbstractStub implements GenericMapper<PactRecord, PactRecord> {
 	

pact/pact-common/src/main/java/eu/stratosphere/pact/common/util/InstantiationUtil.java

@@ -15,9 +15,14 @@
 
 package eu.stratosphere.pact.common.util;
 
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Modifier;
 
+import eu.stratosphere.nephele.configuration.Configuration;
+
 
 /**
  * Utility class to create instances from class objects and checking failure reasons.
@@ -164,6 +169,23 @@ public class InstantiationUtil {
 		}
 	}
 	
+	public static Object readObjectFormConfig(Configuration config, String key) throws IOException, ClassNotFoundException {
+		byte[] bytes = config.getBytes(key, null);
+		if (bytes == null) {
+			return null;
+		}
+		
+		ObjectInputStream oois = null;
+		try {
+			oois = new ObjectInputStream(new ByteArrayInputStream(bytes));
+			return oois.readObject();
+		} finally {
+			if (oois != null) {
+				oois.close();
+			}
+		}
+	}
+	
 	// --------------------------------------------------------------------------------------------
 	
 	/**

pact/pact-common/src/main/java/eu/stratosphere/pact/generic/stub/GenericCoGrouper.java

@@ -22,9 +22,9 @@ import eu.stratosphere.pact.common.stubs.Stub;
 
 public interface GenericCoGrouper<V1, V2, O> extends Stub {
 	
-	public abstract void coGroup(Iterator<V1> records1, Iterator<V2> records2, Collector<O> out);
+	public abstract void coGroup(Iterator<V1> records1, Iterator<V2> records2, Collector<O> out) throws Exception;
 	
-	public abstract void combineFirst(Iterator<V1> records, Collector<V1> out);
+	public void combineFirst(Iterator<V1> records, Collector<V1> out) throws Exception;
 	
-	public abstract void combineSecond(Iterator<V2> records, Collector<V2> out);
+	public void combineSecond(Iterator<V2> records, Collector<V2> out) throws Exception;
 }

pact/pact-common/src/main/java/eu/stratosphere/pact/generic/stub/GenericCrosser.java

@@ -21,5 +21,5 @@ import eu.stratosphere.pact.common.stubs.Stub;
 
 public interface GenericCrosser<V1, V2, O> extends Stub
 {
-	void cross(V1 record1, V2 record2, Collector<O> out);
+	void cross(V1 record1, V2 record2, Collector<O> out) throws Exception;
 }

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

@@ -1083,7 +1083,7 @@ public class TaskConfig {
 		}
 
 		@Override
-		public <T> Class<T> getClass(String key, Class<? extends T> defaultValue, Class<T> ancestor) {
+		public <T> Class<T> getClass(String key, Class<? extends T> defaultValue, Class<? super T> ancestor) {
 			return this.backingConfig.getClass(this.prefix + key, defaultValue, ancestor);
 		}