private static void validateTopic(final String topic, final DittoHeaders dittoHeaders,
        final String placeholderReplacement) {

    if (topic.isEmpty()) {
        throwEmptyException("topic", dittoHeaders);
    }

    try {
        final String topicWithoutPlaceholders =
                topic.replaceAll(Pattern.quote(placeholderReplacement), DUMMY_TOPIC);
        Topic.validate(topicWithoutPlaceholders);
    } catch (final InvalidTopicException e) {
        final String message = MessageFormat.format(INVALID_TOPIC_FORMAT, topic, e.getMessage());
        throw ConnectionConfigurationInvalidException.newBuilder(message)
                .dittoHeaders(dittoHeaders)
                .cause(e)
                .build();
    }
}
 

private static String validateTopic(final String topic) {
    try {
        Topic.validate(topic);
        return topic;
    } catch (final InvalidTopicException e) {
        throw ConnectionConfigurationInvalidException.newBuilder(e.getMessage())
                .cause(e)
                .build();
    }
}
 

@Test(expected = InvalidTopicException.class)
public void createTopic_invalidTopicName() {
    client.createTopic("I'm not a valid topic!", 1, 1);
}
 

/**
 * Simple partitioning on the message key
 * Odd keys to partition 0
 * Even keys to partition 1
 *
 * @param topic topic Name
 * @param key Message Key
 * @param keyBytes Key Bytes
 * @param value Message Value
 * @param valueBytes Value Bytes
 * @param cluster Cluster Object
 * @return Partition Id
 */
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
    if ((keyBytes == null) || (!(key instanceof Integer)))
        throw new InvalidRecordException("Topic Key must have a valid Integer value.");

    if (cluster.partitionsForTopic(topic).size() != 2)
        throw new InvalidTopicException("Topic must have exactly two partitions");

    return (Integer) key % 2;
}