public void replay(TopicRecord record) {
        Uuid existingUuid = topicsByName.put(record.name(), record.topicId());
        if (existingUuid != null) {
            // We don't currently support sending a second TopicRecord for the same topic name...
            // unless, of course, there is a RemoveTopicRecord in between.
            if (existingUuid.equals(record.topicId())) {
                throw new RuntimeException("Found duplicate TopicRecord for " + record.name() +
                        " with topic ID " + record.topicId());
            } else {
                throw new RuntimeException("Found duplicate TopicRecord for " + record.name() +
                        " with a different ID than before. Previous ID was " + existingUuid +
                        " and new ID is " + record.topicId());
            }
        }
        if (Topic.hasCollisionChars(record.name())) {
            String normalizedName = Topic.unifyCollisionChars(record.name());
            TimelineHashSet<String> topicNames = topicsWithCollisionChars.get(normalizedName);
            if (topicNames == null) {
                topicNames = new TimelineHashSet<>(snapshotRegistry, 1);
                topicsWithCollisionChars.put(normalizedName, topicNames);
            }
            topicNames.add(record.name());
        }
        topics.put(record.topicId(),
            new TopicControlInfo(record.name(), snapshotRegistry, record.topicId()));
        log.info("Replayed TopicRecord for topic {} with topic ID {}.", record.name(), record.topicId());
    }
@Test
    public void testDuplicateTopicIdReplay() {
        ReplicationControlTestContext ctx = new ReplicationControlTestContext.Builder().build();
        ReplicationControlManager replicationControl = ctx.replicationControl;
        replicationControl.replay(new TopicRecord().
                setName("foo").
                setTopicId(Uuid.fromString("Ktv3YkMQRe-MId4VkkrMyw")));
        assertEquals("Found duplicate TopicRecord for foo with topic ID Ktv3YkMQRe-MId4VkkrMyw",
            assertThrows(RuntimeException.class,
                () -> replicationControl.replay(new TopicRecord().
                    setName("foo").
                    setTopicId(Uuid.fromString("Ktv3YkMQRe-MId4VkkrMyw")))).
                        getMessage());
        assertEquals("Found duplicate TopicRecord for foo with a different ID than before. " +
            "Previous ID was Ktv3YkMQRe-MId4VkkrMyw and new ID is 8auUWq8zQqe_99H_m2LAmw",
                assertThrows(RuntimeException.class,
                        () -> replicationControl.replay(new TopicRecord().
                                setName("foo").
                                setTopicId(Uuid.fromString("8auUWq8zQqe_99H_m2LAmw")))).
                        getMessage());
    }