An example Kafka encoder implementation for Osso events.

EventEncoder.java

/*
 * Note that this class is a slightly hacked up version of an internal
 * class used at Rocana. There may be some mistakes in the code when I
 * removed some Rocana-specific bits from it. I didn't test that it
 * compiles cleanly.
 */

import kafka.serializer.Encoder;
import kafka.utils.VerifiableProperties;
import org.apache.avro.io.DatumWriter;
import org.apache.avro.io.EncoderFactory;
import org.apache.avro.specific.SpecificData;

import java.io.ByteArrayOutputStream;
import java.io.IOException;

public class EventEncoder implements Encoder<Event> {

  private static final int INITIAL_BUFFER_SIZE = 4096;

  private DatumWriter<Event> eventWriter;
  private ByteArrayOutputStream outputStream;
  private org.apache.avro.io.Encoder encoder;

  public EventEncoder(VerifiableProperties properties) {
    // There's nothing to configure, so properties isn't used. -esammer
    
    eventWriter = SpecificData.get().createDatumWriter(Event.SCHEMA$);
    outputStream = new ByteArrayOutputStream(INITIAL_BUFFER_SIZE);
    encoder = EncoderFactory.get().directBinaryEncoder(outputStream, null);
  }

  @Override
  public byte[] toBytes(Event event) {
    outputStream.reset();

    try {
      eventWriter.write(event, encoder);
      encoder.flush();
    } catch (IOException e) {
      throw new EncoderException(String.format("Unable to encode event:%s - Exception follows.", event), e);
    }
    return outputStream.toByteArray();
  }
}

This code is released under the Apache License version 2.0.

This uses Kafka 0.8.x APIs, but it should be trivial to make it work with later versions.