CollectionInputFormat extends GenericInputFormat<T> implements NonParallelInput { private void writeObject(ObjectOutputStream out) throws IOException { out.defaultWriteObject(); final int size = dataSet.size(); out.writeInt(size); if (size > 0) { DataOutputViewStreamWrapper wrapper = new DataOutputViewStreamWrapper(out); for (T element : dataSet){ serializer.serialize(element, wrapper); } } } CollectionInputFormat(Collection<T> dataSet, TypeSerializer<T> serializer); @Override boolean reachedEnd(); @Override void open(GenericInputSplit split); @Override T nextRecord(T record); @Override String toString(); static void checkCollection(Collection<X> elements, Class<X> viewedAs);  }
@Test public void testSerializationFailure() { try { CollectionInputFormat<ElementType> inFormat = new CollectionInputFormat<ElementType>( Collections.singleton(new ElementType()), new TestSerializer(false, true)); ByteArrayOutputStream buffer = new ByteArrayOutputStream(); ObjectOutputStream out = new ObjectOutputStream(buffer); try { out.writeObject(inFormat); fail("should throw an exception"); } catch (TestException e) { } catch (Exception e) { fail("Exception not properly forwarded"); } } catch (Exception e) { e.printStackTrace(); fail(e.getMessage()); } }