@Override
    public boolean filter(Message msg) {
        if (msg.getSourceInputId() == null) {
            return false;
        }

        for (final Map.Entry<String, String> field : staticFields.getOrDefault(msg.getSourceInputId(), Collections.emptyList())) {
            if (!msg.hasField(field.getKey())) {
                msg.addField(field.getKey(), field.getValue());
            } else {
                LOG.debug("Message already contains field [{}]. Not overwriting.", field.getKey());
            }
        }

        return false;
    }
@Test
    @SuppressForbidden("Executors#newSingleThreadExecutor() is okay for tests")
    public void testFilter() throws Exception {
        Message msg = messageFactory.createMessage("hello", "junit", Tools.nowUTC());
        msg.setSourceInputId("someid");

        when(input.getId()).thenReturn("someid");
        when(inputService.all()).thenReturn(Collections.singletonList(input));
        when(inputService.find(eq("someid"))).thenReturn(input);
        when(inputService.getStaticFields(eq(input)))
                .thenReturn(Collections.singletonList(Maps.immutableEntry("foo", "bar")));

        final StaticFieldFilter filter = new StaticFieldFilter(inputService, new EventBus(), Executors.newSingleThreadScheduledExecutor());
        filter.lifecycleChanged(Lifecycle.STARTING);
        filter.filter(msg);

        assertEquals("hello", msg.getMessage());
        assertEquals("junit", msg.getSource());
        assertEquals("bar", msg.getField("foo"));
    }