@Override
        public synchronized StreamThreadExceptionResponse handle(final Throwable exception) {
            final String topologyName =
                exception instanceof StreamsException && ((StreamsException) exception).taskId().isPresent() ?
                    ((StreamsException) exception).taskId().get().topologyName()
                    : null;

            newErrorsByTopology.computeIfAbsent(topologyName, t -> new LinkedList<>()).add(exception);
            if (exception.getCause() instanceof MissingSourceTopicException) {
                return StreamThreadExceptionResponse.REPLACE_THREAD;
            } else {
                return StreamThreadExceptionResponse.SHUTDOWN_APPLICATION;
            }
        }
@return  true iff all fields other than {@link StreamsMetadataImpl#topologyName()}
         match between the two StreamsMetadata objects
streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java