@Test
public void testFail() throws Exception {

    Map<String, Object> fieldsAndValues = new HashMap<>();
    fieldsAndValues.put("foobar", "100");
    final StreamlineEvent event = StreamlineEventImpl.builder()
            .fieldsAndValues(fieldsAndValues)
            .dataSourceId("srcid")
            .build();
    final Notification notification = new StreamlineEventAdapter(event);
    new MockUp<NotificationQueueHandler>() {
        @Mock
        public void enqueue(Notifier notifier, Notification notification1) {
            notifier.notify(notification);
        }
        @Mock
        public void resubmit(String notificationId) {
            notifier.notify(notification);
        }
    };
    new Expectations() {{
        mockProxyUtil.loadClassFromJar(anyString, "TestClass");
        result = notifier;
        tuple.getValueByField(anyString);
        result = event;
    }};
    Map<String, String> stormConf = new HashMap<>();
    stormConf.put("catalog.root.url", "http://localhost:8080/api/v1/catalog");
    stormConf.put("local.notifier.jar.path", "/tmp");
    bolt.prepare(stormConf, null, collector);
    bolt.execute(tuple);
    new Verifications() {
        {
            hBaseNotificationStore.store(notification);
            times = 1;
            hBaseNotificationStore.updateNotificationStatus(notification.getId(), Notification.Status.FAILED);
            times = 1;
            collector.ack(tuple);
            times = 0;
            collector.fail(tuple);
            times = 1;
        }
    };
}