public void persistMessage(final String topicKey,
                               final String userKey,
                               final String userName,
                               final String message,
                               final int messageNum) throws TopicAccessException {
        assert topicKey != null && topicKey.trim().length() > 0;
        assert userKey != null && userKey.trim().length() > 0;
        assert userName != null && userName.trim().length() > 0;
        assert message != null;

        try {
            new TransactionRunner<Void>(getDatastore()) {
                protected Void txnBlock(Transaction txn) {
                    try {
                        Key subscriberEntityKey = subscriberEntityKey(topicKey, userKey);
                        Entity subscriberEntity = datastore.get(subscriberEntityKey);
                        subscriberEntity.setUnindexedProperty(MESSAGE_NUM_PROP, messageNum);
                        datastore.put(subscriberEntity);
                        txn.commit();

                        if (shouldSaveMessages) {
                            Entity messageEntity = new Entity(MESSAGE_KIND);
                            messageEntity.setProperty(TOPIC_KEY_PROP, topicEntityKey(topicKey));
                            messageEntity.setProperty(USER_KEY_PROP, userKey);
                            messageEntity.setProperty(USER_NAME_PROP, userName);
                            messageEntity.setUnindexedProperty(MESSAGE_TEXT_PROP, new Text(message));
                            messageEntity.setUnindexedProperty(MESSAGE_NUM_PROP, messageNum);
                            datastore.put(messageEntity);
                        }
                    } catch (EntityNotFoundException enfe) {
                        throw new TopicAccessException("Topic " + topicKey + " not found");
                    }
                    return null;
                }
            }.run();
        } catch (Exception e) {
            throw new TopicAccessException("Error persisting message to topic " + topicKey, e);
        }
    }