@Test(expected = SourceException.class)
public void testProcessEventFailure() throws Exception {

  RuntimeException testException = new RuntimeException();
  when(filter.apply(event)).thenThrow(testException);
  source.setStarted(false);
  source.processEvent(event);
  verify(listener, times(0)).onMutation(any());
  verify(metrics, times(0)).eventFailure(testException);
  verify(listener, times(0)).onError(testException);
  source.setStarted(true);
  source.processEvent(event);
  verify(listener, times(0)).onMutation(any());
  verify(metrics, times(1)).eventFailure(testException);
  verify(listener, times(1)).onError(testException);
}