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));

        }

    }

    public void ack(Object msgId) {

        KafkaMessageId id = (KafkaMessageId) msgId;

        PartitionManager m = _coordinator.getManager(id.partition);

        if (m != null) {

            m.ack(id.offset);

        }

    }