public void updateSourceTopics(final Map<String, List<String>> allSourceTopicsByNodeName) {
        sourceNodesByTopic.clear();
        for (final Map.Entry<String, SourceNode<?, ?>> sourceNodeEntry : sourceNodesByName.entrySet()) {
            final String sourceNodeName = sourceNodeEntry.getKey();
            final SourceNode<?, ?> sourceNode = sourceNodeEntry.getValue();

            final List<String> updatedSourceTopics = allSourceTopicsByNodeName.get(sourceNodeName);
            if (updatedSourceTopics == null) {
                log.error("Unable to find source node {} in updated topics map {}",
                          sourceNodeName, allSourceTopicsByNodeName);
                throw new IllegalStateException("Node " + sourceNodeName + " not found in full topology");
            }

            log.trace("Updating source node {} with new topics {}", sourceNodeName, updatedSourceTopics);
            for (final String topic : updatedSourceTopics) {
                if (sourceNodesByTopic.containsKey(topic)) {
                    log.error("Tried to subscribe topic {} to two nodes when updating topics from {}",
                              topic, allSourceTopicsByNodeName);
                    throw new IllegalStateException("Topic " + topic + " was already registered to source node "
                                                        + sourceNodesByTopic.get(topic).name());
                }
                sourceNodesByTopic.put(topic, sourceNode);
            }
        }
    }
@Test
    public void shouldThrowIfMultipleSourceNodeOfSameSubtopologySubscribedToSameTopic() {
        final String sourceNode = "source-1";
        final String updatedSourceNode = "source-2";
        final String doublySubscribedTopic = "topic-1";
        final String topic = "topic-2";
        topology.addSource(sourceNode, doublySubscribedTopic);
        topology.addSource(updatedSourceNode, topic);
        final ProcessorTopology processorTopology = topology.getInternalBuilder("X").buildTopology();

        final Throwable exception = assertThrows(
            IllegalStateException.class,
            () -> processorTopology.updateSourceTopics(mkMap(
                mkEntry(sourceNode, Collections.singletonList(doublySubscribedTopic)),
                mkEntry(updatedSourceNode, Arrays.asList(topic, doublySubscribedTopic))
            ))
        );
        assertThat(
            exception.getMessage(),
            startsWith("Topic " + doublySubscribedTopic + " was already registered to source node")
        );
    }