Bytes prefixBytes(final KO key) {
        //The serialization format. Note that primaryKeySerialized is not required/used in this function.
        //{Integer.BYTES foreignKeyLength}{foreignKeySerialized}{Optional-primaryKeySerialized}

        final byte[] foreignKeySerializedData = foreignKeySerializer.serialize(foreignKeySerdeTopic, key);

        final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + foreignKeySerializedData.length);
        buf.putInt(foreignKeySerializedData.length);
        buf.put(foreignKeySerializedData);
        return Bytes.wrap(buf.array());
    }
@Test
    public void prefixKeySerdeTest() {
        final CombinedKeySchema<String, Integer> cks = new CombinedKeySchema<>(
            () -> "fkTopic", Serdes.String(),
            () -> "pkTopic", Serdes.Integer()
        );
        final String foreignKey = "someForeignKey";
        final byte[] foreignKeySerializedData =
            Serdes.String().serializer().serialize("fkTopic", foreignKey);
        final Bytes prefix = cks.prefixBytes(foreignKey);

        final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + foreignKeySerializedData.length);
        buf.putInt(foreignKeySerializedData.length);
        buf.put(foreignKeySerializedData);
        final Bytes expectedPrefixBytes = Bytes.wrap(buf.array());

        assertEquals(expectedPrefixBytes, prefix);
    }