@Override
  public TimeSeriesKey<K> fromBytes(byte[] timeSeriesKeyBytes) {
    // First obtain the key bytes, and deserialize them. Later de-serialize the timestamp and sequence number
    ByteBuffer buf = ByteBuffer.wrap(timeSeriesKeyBytes);
    int keySize =  timeSeriesKeyBytes.length - TIMESTAMP_SIZE - SEQNUM_SIZE;
    K key = null;

    if (keySize != 0) {
      byte[] keyBytes = new byte[keySize];
      buf.get(keyBytes);
      key = keySerde.fromBytes(keyBytes);
    }

    long timeStamp = buf.getLong();
    long seqNum = buf.getLong();
    long version = seqNum & ~SEQUENCE_NUM_MASK;

    if (version != TimeSeriesKey.VERSION) {
      throw new SamzaException(String.format("Invalid version detected in TimeSeriesKey. " +
          "Expected Version: %s Actual Version: %s Sequence number: %s", TimeSeriesKey.VERSION, version, seqNum));
    }
    return new TimeSeriesKey(key, timeStamp, seqNum);
  }
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