public void generateTuplesWithMessageAndMetadataScheme() {

        String value = "value";

        Partition mockPartition = Mockito.mock(Partition.class);

        mockPartition.partition = 0;

        long offset = 0L;

        

        MessageMetadataSchemeAsMultiScheme scheme = new MessageMetadataSchemeAsMultiScheme(new StringMessageAndMetadataScheme());

        

        createTopicAndSendMessage(null, value);

        ByteBufferMessageSet messageAndOffsets = getLastMessage();

        for (MessageAndOffset msg : messageAndOffsets) {

            Iterable<List<Object>> lists = KafkaUtils.generateTuples(scheme, msg.message(), mockPartition, offset);

            List<Object> values = lists.iterator().next(); 

            assertEquals("Message is incorrect", value, values.get(0));

            assertEquals("Partition is incorrect", mockPartition.partition, values.get(1));

            assertEquals("Offset is incorrect", offset, values.get(2));

        }

    }