EventToLMMSerializerBolt extends BaseRichBolt { @Override public void execute(Tuple tuple) { Event event = (Event) tuple.getValueByField(Constants.FIELD_EVENT); Long ts = (Long) event.getHeaders().get(Constants.FIELD_TIMESTAMP); calendar.setTimeInMillis(ts); event.getHeaders().put(TIMESTAMP, formatter.format(calendar.getTime())); event.getHeaders().put(TENANT_ID, event.getHeaders().get(Constants.FIELD_RULE_GROUP)); String eventJson = gson.toJson(event.getHeaders()); collector.emit(tuple, new Values(event.getHeaders().get(TENANT_ID), eventJson)); collector.ack(tuple); }  @SuppressWarnings("rawtypes") @Override void prepare(Map stormConf, TopologyContext context, OutputCollector collector); @Override void execute(Tuple tuple); @Override void declareOutputFields(OutputFieldsDeclarer declarer); static final String TIMESTAMP; static final String TENANT_ID; }
@Test public void testExecute() { EventToLMMSerializerBolt bolt = new EventToLMMSerializerBolt(); Event event = new UnifiedFactory().buildEvent(); final AtomicReference<Values> outputTuple = new AtomicReference<Values>(null); OutputCollector mockCollector = MockTupleHelpers.mockCollector(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Object newEvent = invocation.getArguments()[1]; outputTuple.set((Values) newEvent); return new ArrayList<>(); } }); Calendar instance = Calendar.getInstance(TimeZone.getTimeZone("UTC")); instance.setTimeInMillis(1478736000000L); System.out.println("Input date:"+instance.getTime().getTime()); long ts = instance.getTimeInMillis(); String fakeTenant = UUID.randomUUID().toString().replace("-", ""); event.getHeaders().put(Constants.FIELD_TIMESTAMP, ts); event.getHeaders().put(Constants.FIELD_RULE_GROUP, fakeTenant); when(tuple.getValueByField(Constants.FIELD_EVENT)).thenReturn(event); Map<String, String> conf = new HashMap<>(); bolt.prepare(conf, null, mockCollector); bolt.execute(tuple); verify(mockCollector, times(1)).ack(tuple); Values values = outputTuple.get(); assertEquals(fakeTenant, values.get(0)); JsonObject json = new Gson().fromJson((String) values.get(1), JsonObject.class); assertEquals("2016-11-10T00:00:00.000Z", json.get(EventToLMMSerializerBolt.TIMESTAMP).getAsString()); assertEquals(fakeTenant, json.get(EventToLMMSerializerBolt.TENANT_ID).getAsString()); mockCollector = MockTupleHelpers.mockCollector(new Answer<Object>() { @Override public Object answer(InvocationOnMock invocation) throws Throwable { Object newEvent = invocation.getArguments()[1]; outputTuple.set((Values) newEvent); return new ArrayList<>(); } }); when(tuple.getString(0)).thenReturn((String) values.get(1)); InterceptionBolt interceptor = new InterceptionBolt(); interceptor.prepare(conf, null, mockCollector); interceptor.execute(tuple); verify(mockCollector, times(1)).ack(tuple); json = new Gson().fromJson((String) outputTuple.get().get(0), JsonObject.class); System.out.println(outputTuple.get().get(0)); assertEquals(ts, json.get("@timestamp").getAsLong()); }