Bytes toBytes(final KO foreignKey, final K primaryKey) {
        //The serialization format - note that primaryKeySerialized may be null, such as when a prefixScan
        //key is being created.
        //{Integer.BYTES foreignKeyLength}{foreignKeySerialized}{Optional-primaryKeySerialized}
        final byte[] foreignKeySerializedData = foreignKeySerializer.serialize(foreignKeySerdeTopic,
                                                                               foreignKey);

        //? bytes
        final byte[] primaryKeySerializedData = primaryKeySerializer.serialize(primaryKeySerdeTopic,
                                                                               primaryKey);

        final ByteBuffer buf = ByteBuffer.allocate(Integer.BYTES + foreignKeySerializedData.length + primaryKeySerializedData.length);
        buf.putInt(foreignKeySerializedData.length);
        buf.put(foreignKeySerializedData);
        buf.put(primaryKeySerializedData);
        return Bytes.wrap(buf.array());
    }
@Test
    public void nullPrimaryKeySerdeTest() {
        final CombinedKeySchema<String, Integer> cks = new CombinedKeySchema<>(
            () -> "fkTopic", Serdes.String(),
            () -> "pkTopic", Serdes.Integer()
        );
        assertThrows(NullPointerException.class, () -> cks.toBytes("foreignKey", null));
    }