@Override public Writable serialize(Object obj, ObjectInspector objInspector) throws SerDeException {

    if (!(objInspector instanceof StructObjectInspector)) {
      throw new SerDeException("Object inspector has to be "
          + StructObjectInspector.class.getName()
          + " but got "
          + objInspector.getClass().getName());
    }
    StructObjectInspector structObjectInspector = (StructObjectInspector) objInspector;
    List<Object> data = structObjectInspector.getStructFieldsDataAsList(obj);
    int firstMetadataColumnIndex = data.size() - MetadataColumn.values().length;
    if (delegateSerializerOI == null) {
      //@TODO check if i can cache this if it is the same.
      delegateSerializerOI = new SubStructObjectInspector(structObjectInspector, firstMetadataColumnIndex);
    }
    // We always append the metadata columns to the end of the row.
    final List<Object> row = data.subList(0, firstMetadataColumnIndex);

    Object key = data.get(firstMetadataColumnIndex);
    Object partition = data.get(firstMetadataColumnIndex + 1);
    Object offset = data.get(firstMetadataColumnIndex + 2);
    Object timestamp = data.get(firstMetadataColumnIndex + 3);

    if (PrimitiveObjectInspectorUtils.getLong(offset, MetadataColumn.OFFSET.getObjectInspector()) != -1) {
      throw new SerDeException("Can not insert values into `__offset` column, has to be [-1]");
    }

    final byte[]
        keyBytes =
        key == null ? null : PrimitiveObjectInspectorFactory.writableBinaryObjectInspector.getPrimitiveJavaObject(key);
    final long
        recordTs =
        timestamp == null ?
            -1 :
            PrimitiveObjectInspectorUtils.getLong(timestamp, MetadataColumn.TIMESTAMP.getObjectInspector());
    final int
        recordPartition =
        partition == null ?
            -1 :
            PrimitiveObjectInspectorUtils.getInt(partition, MetadataColumn.PARTITION.getObjectInspector());

    //noinspection unchecked
    return new KafkaWritable(recordPartition,
        recordTs,
        bytesConverter.getBytes(delegateSerDe.serialize(row, delegateSerializerOI)),
        keyBytes);
  }
Object Inspector of original row plus metadata.
kafka-handler/src/java/org/apache/hadoop/hive/kafka/KafkaSerDe.java