AbstractSource extends ListenableSource<E> { @Override public final void open() { try { if (isStarted()) { log.info("Source {} already started", name); return; } Preconditions.checkState( isTerminated(), "Previous processor thread has not terminated for source %s", name); initialize(); notifyStart(); started.set(true); start(); log.info("Started source {}", name); metrics.start(); } catch (Throwable ex) { final String errorMessage = String.format("Failed to start source %s", name); log.error(errorMessage, ex); metrics.startFailure(ex); close(); throw new SourceException(errorMessage, ex); } }  @Override final void open(); @Override final void close(); @Override final void checkpoint(Mutation<?> mutation); @Override void clear(); final void processEvent(final E event); final void addEventValidator(@NonNull final Validator<E> validator); final void addMutationValidator(
      @NonNull final Validator<M> validator);  }
@Test(expected = SourceException.class) public void testOpenWhilePreviousProcessorRunning() throws Exception { source.open(); source.setStarted(false); source.open(); }