@Override
  public byte[] toBytes(TimeSeriesKey<K> timeSeriesKey) {
    K key = timeSeriesKey.getKey();
    long timestamp = timeSeriesKey.getTimestamp();
    long seqNum = timeSeriesKey.getSeqNum();

    byte[] serializedKey = null;
    if (keySerde != null) {
      serializedKey = keySerde.toBytes(key);
    }
    int keySize = serializedKey == null ? 0 : serializedKey.length;

    // append the timestamp and sequence number to the serialized key bytes
    ByteBuffer buf = ByteBuffer.allocate(keySize + TIMESTAMP_SIZE + SEQNUM_SIZE);
    if (serializedKey != null) {
      buf.put(serializedKey);
    }
    buf.putLong(timestamp);
    buf.putLong(seqNum & SEQUENCE_NUM_MASK);

    return buf.array();
  }
A {@link Serde} for {@link TimeSeriesKey}s.

<p>
This wraps the actual key's serde with serializers for timestamp, version number and sequence number.

A {@link TimeSeriesKeySerde} serializes a key as follows:
   +-------------------------+------------------+----------------+------------------+
   |  serialized-key bytes   |  timestamp       | version (0)    | seqNum           |
   |(serialized by keySerde) |                  |                |                  |
   +-------------------------+------------------+----------------+------------------+
   +---serialized key len----+-------8 bytes----+---1 byte-------+---7 bytes---------+

@param <K> the type of the wrapped key
samza-core/src/main/java/org/apache/samza/operators/impl/store/TimeSeriesKeySerde.java