code: public static ByteBuffer toSerializedEvent(AbstractEvent event) throws IOException {
  		final Class<?> eventClass = event.getClass();
  		if (eventClass == EndOfPartitionEvent.class) {
  			return ByteBuffer.wrap(new byte[] { 0, 0, 0, END_OF_PARTITION_EVENT });
  		}
  		else if (eventClass == CheckpointBarrier.class) {
  			return serializeCheckpointBarrier((CheckpointBarrier) event);
  		}
  		else if (eventClass == EndOfSuperstepEvent.class) {
  			return ByteBuffer.wrap(new byte[] { 0, 0, 0, END_OF_SUPERSTEP_EVENT });
  		}
  		else if (eventClass == CancelCheckpointMarker.class) {
  			CancelCheckpointMarker marker = (CancelCheckpointMarker) event;
  
  			ByteBuffer buf = ByteBuffer.allocate(12);
  			buf.putInt(0, CANCEL_CHECKPOINT_MARKER_EVENT);
  			buf.putLong(4, marker.getCheckpointId());
  			return buf;
  		}
  		else {
  			try {
  				final DataOutputSerializer serializer = new DataOutputSerializer(128);
  				serializer.writeInt(OTHER_EVENT);
  				serializer.writeUTF(event.getClass().getName());
  				event.write(serializer);
  				return serializer.wrapAsByteBuffer();
  			}
  			catch (IOException e) {
  				throw new IOException("Error while serializing event.", e);
  			}
  		}
  	}
docs: ------------------------------------------------------------------------