diff --git a/lang/common/wake/avro/AvroTestMessage.avsc b/lang/common/wake/avro/AvroTestMessage.avsc
new file mode 100644
index 0000000..403d900
--- /dev/null
+++ b/lang/common/wake/avro/AvroTestMessage.avsc
@@ -0,0 +1,41 @@
+﻿/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+[
+  /*
+   * Identify the next message in the Java/C# bridge protocol.
+   */
+  {
+    "namespace":"org.apache.reef.wake.test.avro.message",
+    "type":"record",
+    "name":"AvroTestMessage",
+    "doc":"Identifies the following message in a given protocol.",
+    "fields":[
+      {
+        "name":"number",
+        "doc":"Integer test field.",
+        "type":"int"
+      },
+      {
+        "name":"data",
+        "doc":"String test field.",
+        "type":"string"
+      }
+    ]
+  }
+]
diff --git a/lang/common/wake/avro/Header.avsc b/lang/common/wake/avro/Header.avsc
index 8efd4f5..b2882b3 100644
--- a/lang/common/wake/avro/Header.avsc
+++ b/lang/common/wake/avro/Header.avsc
@@ -1,4 +1,4 @@
-﻿/*
+/*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
diff --git a/lang/java/reef-wake/wake/pom.xml b/lang/java/reef-wake/wake/pom.xml
index 7f1cfee..cdb82dc 100644
--- a/lang/java/reef-wake/wake/pom.xml
+++ b/lang/java/reef-wake/wake/pom.xml
@@ -52,6 +52,22 @@
         </resources>
 
         <plugins>
+           <plugin>
+              <groupId>org.apache.avro</groupId>
+              <artifactId>avro-maven-plugin</artifactId>
+              <executions>
+                <execution>
+                    <phase>generate-sources</phase>
+                    <goals>
+                        <goal>schema</goal>
+                    </goals>
+                    <configuration>
+                        <sourceDirectory>${rootPath}/lang/common/wake/avro/</sourceDirectory>
+                        <outputDirectory>${project.basedir}/target/generated-sources/avro/</outputDirectory>
+                    </configuration>
+                </execution>
+              </executions>
+            </plugin>
             <plugin>
                 <artifactId>maven-antrun-plugin</artifactId>
                 <executions>
@@ -151,6 +167,15 @@
             <groupId>net.jcip</groupId>
             <artifactId>jcip-annotations</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.github.lukehutch</groupId>
+            <artifactId>fast-classpath-scanner</artifactId>
+            <version>${fast-classpath-scanner.version}</version>
+        </dependency>
     </dependencies>
 
 </project>
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/MultiObserver.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/MultiObserver.java
new file mode 100644
index 0000000..e310439
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/MultiObserver.java
@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake;
+
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * MultiObserver interface definition whose implementation allows the
+ * a single java class to be the recipient of multiple events by simply
+ * defining the methods to handle those events.
+ */
+public interface MultiObserver {
+  /**
+   * Generic event onNext method in the base interface which maps the call to a concrete
+   * event onNext method in TSubCls if one exists otherwise unimplemented is invoked.
+   * @param event An event of type TEvent which will be sent to TSubCls as appropriate.
+   * @param <TEvent> The type of the event being processed.
+   */
+  <TEvent> void onNext(long identifier, TEvent event) throws IllegalAccessException, InvocationTargetException;
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageDeserializer.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageDeserializer.java
new file mode 100644
index 0000000..e918d90
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageDeserializer.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake.avro;
+
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.reef.wake.MultiObserver;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Base interface for Avro message deserializer objects.
+ */
+public interface IMessageDeserializer {
+  /**
+   * Deserialize messages of type TMessage from input decoder.
+   * @param decoder An Avro BinaryDecoder instance that is reading the input stream containing the message.
+   * @param observer An instance of the MultiObserver class that will process the message.
+   * @param sequence A long value which contains the sequence number of the message in the input stream.
+   * @throws IOException Read of input stream in decoder failed.
+   * @throws IllegalAccessException Target method in observer is not accessible.
+   * @throws InvocationTargetException Subclass threw and exception.
+   */
+  void deserialize(BinaryDecoder decoder, MultiObserver observer, long sequence)
+   throws IOException, IllegalAccessException, InvocationTargetException;
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageSerializer.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageSerializer.java
new file mode 100644
index 0000000..9c1e071
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/IMessageSerializer.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake.avro;
+
+import org.apache.avro.specific.SpecificRecord;
+
+import java.io.IOException;
+import java.io.ByteArrayOutputStream;
+
+/**
+ * Base interface for Avro message serializer objects.
+ */
+public interface IMessageSerializer {
+  /**
+   * Deserialize messages of type TMessage from input outputStream.
+   * @param outputStream A ByteArrayOutputStream where the message to
+   *                     be serialized will be written.
+   * @param message An Avro message class which implements the Avro SpcificRecord interface.
+   * @param sequence The numerical position of the message in the outgoing message stream.
+   * @throws IOException An error occurred writing the message to the outputStream.
+   */
+  void serialize(ByteArrayOutputStream outputStream, SpecificRecord message, long sequence)
+    throws IOException;
+}
+
+
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/ProtocolSerializer.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/ProtocolSerializer.java
new file mode 100644
index 0000000..ad10d5a
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/ProtocolSerializer.java
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake.avro;
+
+import org.apache.avro.io.*;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.avro.specific.SpecificRecord;
+import org.apache.avro.specific.SpecificRecordBase;
+import org.apache.reef.wake.MultiObserver;
+import org.apache.reef.wake.avro.message.Header;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.InputStream;
+import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+import io.github.lukehutch.fastclasspathscanner.FastClasspathScanner;
+import io.github.lukehutch.fastclasspathscanner.scanner.ScanResult;
+
+/**
+ * The ProtocolSerializer generates serializers and deserializers for
+ * all of the Avro messages contained in a specified package. The name
+ * of the package must have "message" as the final component of the
+ * package name. For example, Avro messages in the org.foo.me package
+ * would sit in the org.foo.me.messages package.
+ */
+public final class ProtocolSerializer {
+  private static final Logger LOG = Logger.getLogger(ProtocolSerializer.class.getName());
+  // Maps for mapping message class names to serializer and deserializer classes.
+  private final Map<String, IMessageSerializer> nameToSerializerMap = new HashMap<>();
+  private final Map<String, IMessageDeserializer> nameToDeserializerMap = new HashMap<>();
+  private final SpecificDatumReader<Header> headerReader = new SpecificDatumReader<>(Header.class);
+
+  /**
+   * Finds all of the messages in the specified packaged and calls register.
+   * @param messagePackage A string which contains the full name of the
+   *                       package containing the protocol messages.
+   */
+  public ProtocolSerializer(final String messagePackage) {
+    // Build a list of the message reflection classes.
+    final ScanResult scanResult = new FastClasspathScanner(messagePackage).scan();
+    final List<String> scanNames = scanResult.getNamesOfSubclassesOf(SpecificRecordBase.class);
+    final List<Class<?>> messageClasses = scanResult.classNamesToClassRefs(scanNames);
+
+    // Add the header message from the org.apache.reef.wake.avro.message package.
+    messageClasses.add(Header.class);
+
+    try {
+      // Register all of the messages in the specified package.
+      for (final Class<?> cls : messageClasses) {
+        this.register(cls);
+      }
+    } catch (final Exception e) {
+      throw new RuntimeException("Message registration failed", e);
+    }
+  }
+
+  /**
+   * Instantiates and adds a message serializer/deserializer for the message.
+   * @param msgMetaClass The reflection class for the message.
+   * @param <TMessage> The Java type of the message being registered.
+   */
+  public <TMessage> void register(final Class<TMessage> msgMetaClass) {
+    LOG.log(Level.INFO, "Registering message: {0}", msgMetaClass.getSimpleName());
+    nameToSerializerMap.put(msgMetaClass.getSimpleName(), SerializationFactory.createSerializer(msgMetaClass));
+    nameToDeserializerMap.put(msgMetaClass.getSimpleName(), SerializationFactory.createDeserializer(msgMetaClass));
+  }
+
+  /**
+   * Marshall the input message to a byte array.
+   * @param message The message to be marshaled into a byte array.
+   * @param sequence The unique sequence number of the message.
+   */
+  public byte[] write(final SpecificRecord message, final long sequence) {
+    try (final ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) {
+      final String name = message.getClass().getSimpleName();
+      LOG.log(Level.FINE, "Serializing message: {0}", name);
+
+      final IMessageSerializer serializer = nameToSerializerMap.get(name);
+      if (serializer != null) {
+        serializer.serialize(outputStream, message, sequence);
+      }
+
+      return outputStream.toByteArray();
+    } catch (final Exception e) {
+      throw new RuntimeException("Failure writing message: " + message.getClass().getCanonicalName(), e);
+    }
+  }
+
+  /**
+   * Read a message from the input byte stream and send it to the event handler.
+   * @param messageBytes An array of bytes that contains the message to be deserialized.
+   * @param observer An implementation of the MultiObserver interface which will be called
+   *                 to process the deserialized message.
+   */
+  public void read(final byte[] messageBytes, final MultiObserver observer) {
+    try (final InputStream inputStream = new ByteArrayInputStream(messageBytes)) {
+      // Binary decoder for both the header and the message.
+      final BinaryDecoder decoder = DecoderFactory.get().binaryDecoder(inputStream, null);
+
+      // Read the header message.
+      final Header header = headerReader.read(null, decoder);
+      LOG.log(Level.FINE, "Deserializing Avro message: {0}", header.getClassName());
+
+      // Get the appropriate deserializer and deserialize the message.
+      final IMessageDeserializer deserializer = nameToDeserializerMap.get(header.getClassName().toString());
+      if (deserializer != null) {
+        deserializer.deserialize(decoder, observer, header.getSequence());
+      } else {
+        throw new RuntimeException("Request to deserialize unknown message type: " +  header.getClassName());
+      }
+
+    } catch (final Exception e) {
+      throw new RuntimeException("Failure reading message: ", e);
+    }
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/SerializationFactory.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/SerializationFactory.java
new file mode 100644
index 0000000..be3031d
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/SerializationFactory.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake.avro;
+
+import org.apache.reef.wake.avro.impl.MessageSerializerImpl;
+import org.apache.reef.wake.avro.impl.MessageDeserializerImpl;
+
+/**
+ * Provides Avro message specific message serializers and deserializers.
+ */
+public final class SerializationFactory {
+
+  private SerializationFactory() {}
+
+  /**
+   * Instantiate an Avro message serializer for the message type in the generic parameter.
+   * @param msgMetaClass The reflection class for the message.
+   * @param <TMessage> The type of the Avro message to be serialized.
+   * @return A reference to an IMessageSerializer interface.
+   */
+  public static <TMessage> IMessageSerializer createSerializer(final Class<TMessage> msgMetaClass) {
+    return new MessageSerializerImpl<>(msgMetaClass);
+  }
+
+  /**
+   * Instantiate an Avro message deserializer for the message type in the generic parameter.
+   * @param msgMetaClass The reflection class for the message.
+   * @param <TMessage> The type of the Avro message to be deserialized.
+   * @return A reference to an IMessageDeserializer interface.
+   */
+  public static <TMessage> IMessageDeserializer createDeserializer(final Class<TMessage> msgMetaClass) {
+    return new MessageDeserializerImpl<>(msgMetaClass);
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageDeserializerImpl.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageDeserializerImpl.java
new file mode 100644
index 0000000..55d5716
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageDeserializerImpl.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake.avro.impl;
+
+import org.apache.avro.io.BinaryDecoder;
+import org.apache.avro.specific.SpecificDatumReader;
+import org.apache.reef.wake.MultiObserver;
+import org.apache.reef.wake.avro.IMessageDeserializer;
+
+import java.io.IOException;
+import java.lang.reflect.InvocationTargetException;
+
+/**
+ * Implementation of Avro message specific deserializer.
+ * @param <TMessage> Java type of the message the instantiation can deserialize.
+ */
+public class MessageDeserializerImpl<TMessage> implements IMessageDeserializer {
+  protected Class<TMessage> msgMetaClass;
+  private final SpecificDatumReader<TMessage> messageReader;
+
+  /**
+   * Initialize message specific deserializer.
+   * @param msgMetaClass The reflection class for the message.
+   * @param msgMetaClass
+   */
+  public MessageDeserializerImpl(final Class<TMessage> msgMetaClass) {
+    this.msgMetaClass = msgMetaClass;
+    this.messageReader = new SpecificDatumReader<>(msgMetaClass);
+  }
+
+  /**
+   * Deserialize messages of type TMessage from input decoder.
+   * @param decoder An Avro BinaryDecoder instance that is reading the input stream containing the message.
+   * @param observer An instance of the MultiObserver class that will process the message.
+   * @param sequence A long value which contains the sequence number of the message in the input stream.
+   * @throws IOException Read of input stream in decoder failed.
+   * @throws IllegalAccessException Target method in observer is not accessible.
+   * @throws InvocationTargetException Subclass threw and exception.
+   */
+  public void deserialize(final BinaryDecoder decoder, final MultiObserver observer, final long sequence)
+    throws IOException, IllegalAccessException, InvocationTargetException {
+
+    final TMessage message = messageReader.read(null, decoder);
+    if (message != null) {
+      observer.onNext(sequence, message);
+    } else {
+      throw new RuntimeException("Failed to deserialize message [" + msgMetaClass.getSimpleName() + "]");
+    }
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageSerializerImpl.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageSerializerImpl.java
new file mode 100644
index 0000000..f225cbb
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/MessageSerializerImpl.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.reef.wake.avro.impl;
+
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.DatumWriter;
+import org.apache.avro.io.EncoderFactory;
+import org.apache.avro.specific.SpecificDatumWriter;
+import org.apache.avro.specific.SpecificRecord;
+import org.apache.reef.wake.avro.IMessageSerializer;
+import org.apache.reef.wake.avro.message.Header;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+/**
+ * Implementation of Avro message specific serializer.
+ * @param <TMessage> Java type of the message the instantiation can serialize.
+ */
+public class MessageSerializerImpl<TMessage> implements IMessageSerializer {
+  private final String msgMetaClassName;
+  // Writers for header and message.
+  private final DatumWriter<Header> headerWriter = new SpecificDatumWriter<>(Header.class);
+  private final DatumWriter<TMessage> messageWriter;
+
+  /**
+   * Initialize message specific serializer.
+   * @param msgMetaClass The reflection class for the message.
+   */
+  public MessageSerializerImpl(final Class<TMessage> msgMetaClass) {
+    this.msgMetaClassName = msgMetaClass.getSimpleName();
+    this.messageWriter = new SpecificDatumWriter<>(msgMetaClass);
+  }
+
+  /**
+   * Deserialize messages of type TMessage from input outputStream.
+   * @param outputStream A ByteArrayOutputStream where the message to
+   *                     be serialized will be written.
+   * @param message An Avro message class which implements the Avro SpcificRecord interface.
+   * @param sequence The numerical position of the message in the outgoing message stream.
+   * @throws IOException An error occurred writing the message to the outputStream.
+   */
+  public void serialize(final ByteArrayOutputStream outputStream,
+                        final SpecificRecord message, final long sequence) throws IOException {
+    // Binary encoder for both the header and message.
+    final BinaryEncoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);
+    // Write the header and the message.
+    headerWriter.write(new Header(sequence, msgMetaClassName), encoder);
+    messageWriter.write((TMessage)message, encoder);
+    encoder.flush();
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/package-info.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/package-info.java
new file mode 100644
index 0000000..a10a919
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/impl/package-info.java
@@ -0,0 +1,23 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * Implementations of serializer and derserializer interfaces.
+ */
+package org.apache.reef.wake.avro.impl;
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/package-info.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/package-info.java
new file mode 100644
index 0000000..9423ed9
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/avro/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/**
+ * Wake avro serialization classes.
+ */
+package org.apache.reef.wake.avro;
diff --git a/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/MultiObserverImpl.java b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/MultiObserverImpl.java
new file mode 100644
index 0000000..fb3935d
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/main/java/org/apache/reef/wake/impl/MultiObserverImpl.java
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.reef.wake.impl;
+
+import org.apache.reef.wake.MultiObserver;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * The MultiObserverImpl class uses reflection to discover which onNext()
+ * event processing methods are defined and then map events to them.
+ * @param <TSubCls> The subclass derived from MultiObserverImpl.
+ */
+public abstract class MultiObserverImpl<TSubCls> implements MultiObserver {
+  private static final Logger LOG = Logger.getLogger(MultiObserverImpl.class.getName());
+  private final Map<String, Method> methodMap = new HashMap<>();
+
+  /**
+   * Use reflection to discover all of the event processing methods in TSubCls
+   * and setup a means to direct calls from the generic event onNext method defined
+   * in the MultiObserver interface to specific concrete event onNext methods.
+   */
+  public MultiObserverImpl() {
+    // Iterate across the methods and build a hash map of class names to reflection methods.
+    for (final Method method : this.getClass().getMethods()) {
+      if (method.getName().equals("onNext") && method.getDeclaringClass().equals(this.getClass())) {
+        // This is an onNext method defined in TSubCls
+        final Class<?>[] types = method.getParameterTypes();
+        if (types.length == 2 && types[0].equals(Long.TYPE)) {
+          methodMap.put(types[1].getName(), method);
+        }
+      }
+    }
+  }
+
+  /**
+   * Called when an event is received that does not have an onNext method definition
+   * in TSubCls. Override in TSubClas to handle the error.
+   * @param event A reference to an object which is an event not handled by TSubCls.
+   * @param <TEvent> The type of the event being processed.
+   */
+  private <TEvent> void unimplemented(final long identifier, final TEvent event) {
+    LOG.log(Level.INFO, "Unimplemented event: [{0}]: {1}",
+        new String[]{String.valueOf(identifier), event.getClass().getName()});
+  }
+
+  /**
+   * Generic event onNext method in the base interface which maps the call to a concrete
+   * event onNext method in TSubCls if one exists otherwise unimplemented is invoked.
+   * @param event An event of type TEvent which will be sent to TSubCls as appropriate.
+   * @param <TEvent> The type of the event being processed.
+   */
+  @Override
+  public <TEvent> void onNext(final long identifier, final TEvent event)
+    throws IllegalAccessException, InvocationTargetException {
+
+    // Get the reflection method for this call.
+    final Method onNext = methodMap.get(event.getClass().getName());
+    if (onNext != null) {
+      // Process the event.
+      onNext.invoke((TSubCls) this, identifier, event);
+    } else {
+      // Log the unprocessed event.
+      unimplemented(identifier, event);
+    }
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/ProtocolSerializerTest.java b/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/ProtocolSerializerTest.java
new file mode 100644
index 0000000..31d142a
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/ProtocolSerializerTest.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.reef.wake.test.avro;
+
+import org.apache.reef.tang.Injector;
+import org.apache.reef.tang.Tang;
+import org.apache.reef.tang.exceptions.InjectionException;
+import org.apache.reef.wake.EventHandler;
+import org.apache.reef.wake.avro.ProtocolSerializer;
+import org.apache.reef.wake.impl.LoggingEventHandler;
+import org.apache.reef.wake.impl.MultiObserverImpl;
+import org.apache.reef.wake.remote.*;
+import org.apache.reef.wake.remote.address.LocalAddressProvider;
+import org.apache.reef.wake.remote.impl.ByteCodec;
+import org.apache.reef.wake.remote.ports.TcpPortProvider;
+import org.apache.reef.wake.test.avro.message.AvroTestMessage;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TestName;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Logger;
+
+import static org.junit.Assert.assertEquals;
+
+/**
+ *  Verify the protocol serializer can serialize and deserialize messages
+ *  exchanged between two remote manager classes.
+ */
+public final class ProtocolSerializerTest {
+  private static final Logger LOG = Logger.getLogger(ProtocolSerializer.class.getName());
+
+  @Rule
+  public final TestName name = new TestName();
+
+  /**
+   * Verify Avro message can be serialized and deserialized
+   * between two remote managers.
+   */
+  @Test
+  public void testProtocolSerializerTest() throws Exception {
+    final int[] numbers = {12, 25};
+    final String[] strings = {"The first string", "The second string"};
+
+    // Queues for storing messages byte messages.
+    final BlockingQueue<byte[]> queue1 = new LinkedBlockingQueue<>();
+    final BlockingQueue<byte[]> queue2 = new LinkedBlockingQueue<>();
+
+    // Remote managers for sending and receiving byte messages.
+    final RemoteManager remoteManager1 = getTestRemoteManager("RemoteManagerOne");
+    final RemoteManager remoteManager2 = getTestRemoteManager("RemoteManagerTwo");
+
+    // Register message handlers for byte level messages.
+    remoteManager1.registerHandler(byte[].class, new ByteMessageObserver(queue1));
+    remoteManager2.registerHandler(byte[].class, new ByteMessageObserver(queue2));
+
+    final EventHandler<byte[]> sender1 = remoteManager1.getHandler(remoteManager2.getMyIdentifier(), byte[].class);
+    final EventHandler<byte[]> sender2 = remoteManager2.getHandler(remoteManager1.getMyIdentifier(), byte[].class);
+
+    final ProtocolSerializer serializer = new ProtocolSerializer("org.apache.reef.wake.test.avro.message");
+
+    sender1.onNext(serializer.write(new AvroTestMessage(numbers[0], strings[0]), 1));
+    sender2.onNext(serializer.write(new AvroTestMessage(numbers[1], strings[1]), 2));
+
+    final AvroMessageObserver avroObserver1 = new AvroMessageObserver();
+    final AvroMessageObserver avroObserver2 = new AvroMessageObserver();
+
+    serializer.read(queue1.take(), avroObserver1);
+    serializer.read(queue2.take(), avroObserver2);
+
+    assertEquals(numbers[0], avroObserver2.getNumber());
+    assertEquals(strings[0], avroObserver2.getDataString());
+
+    assertEquals(numbers[1], avroObserver1.getNumber());
+    assertEquals(strings[1], avroObserver1.getDataString());
+  }
+
+  /**
+   * Build a remote manager on the local IP address with an unused port.
+   * @param identifier The identifier of the remote manager.
+   * @return A RemoteManager instance listing on the local IP address
+   *         with a unique port number.
+   */
+  private RemoteManager getTestRemoteManager(final String identifier) throws InjectionException {
+    final int port = 0;
+    final boolean order = true;
+    final int retries = 3;
+    final int timeOut = 10000;
+
+    final Injector injector = Tang.Factory.getTang().newInjector();
+    final LocalAddressProvider localAddressProvider = injector.getInstance(LocalAddressProvider.class);
+    final TcpPortProvider tcpPortProvider = injector.getInstance(TcpPortProvider.class);
+    final RemoteManagerFactory remoteManagerFactory = injector.getInstance(RemoteManagerFactory.class);
+
+    return remoteManagerFactory.getInstance(
+    identifier, localAddressProvider.getLocalAddress(), port, new ByteCodec(),
+      new LoggingEventHandler<Throwable>(), order, retries, timeOut,
+      localAddressProvider, tcpPortProvider);
+  }
+
+  private final class ByteMessageObserver implements EventHandler<RemoteMessage<byte[]>> {
+    private final BlockingQueue<byte[]> queue;
+
+    /**
+     * @param queue Queue where incoming messages will be stored.
+     */
+    ByteMessageObserver(final BlockingQueue<byte[]> queue) {
+      this.queue = queue;
+    }
+
+    /**
+     * Deserialize and direct incoming messages to the registered MuiltiObserver event handler.
+     * @param message A RemoteMessage<byte[]> object which will be deserialized.
+     */
+    public void onNext(final RemoteMessage<byte[]> message) {
+      queue.add(message.getMessage());
+    }
+  }
+
+  /**
+   * Processes messages from the network remote manager.
+   */
+  public final class AvroMessageObserver extends MultiObserverImpl<AvroMessageObserver> {
+    private int number;
+    private String dataString;
+
+    // Accessors
+    int getNumber() {
+      return number;
+    }
+
+    String getDataString() {
+      return dataString;
+    }
+
+    /**
+     * Processes protocol messages from the C# side of the bridge.
+     * @param identifier A long value which is the unique message identifier.
+     * @param message A reference to the received avro test message.
+     */
+    public void onNext(final long identifier, final AvroTestMessage message) {
+      number = message.getNumber();
+      dataString = message.getData().toString();
+    }
+  }
+}
diff --git a/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/package-info.java b/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/package-info.java
new file mode 100644
index 0000000..63744c7
--- /dev/null
+++ b/lang/java/reef-wake/wake/src/test/java/org/apache/reef/wake/test/avro/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+/**
+ * Tests for org.apache.reef.wake.avro package.
+ */
+package org.apache.reef.wake.test.avro;
diff --git a/pom.xml b/pom.xml
index 02ae63f..9d7ee82 100644
--- a/pom.xml
+++ b/pom.xml
@@ -65,6 +65,7 @@
         <jsr305.version>3.0.1</jsr305.version>
         <kryo.version>3.0.3</kryo.version>
         <kryo-serializers.version>0.37</kryo-serializers.version>
+        <fast-classpath-scanner.version>2.4.5</fast-classpath-scanner.version>
         <rootPath>${user.dir}</rootPath>
     </properties>