FlattenJsonBolt extends AbstractLogRecordBolt { @Override protected void execute(Tuple input, RecordCollector collector) { byte[] bytes = (byte[]) input.getValue(0); String record = new String(bytes); String sha1Checksum = DigestUtils.shaHex(bytes); if (StringUtils.isBlank(record)) { logger.info("received blank record"); return; } try { LogRecord logRecord = new LogRecord(sha1Checksum); logRecord.addMetadataValue(SHA1_CHECKSUM, sha1Checksum); parseJson(record, logRecord); collector.emit(logRecord); } catch (Exception e) { logger.error("Failed to process tuple: " + record, e); } }  @Override void configure(Configuration conf);  }
@Test public void testExecute( @Injectable Tuple input, @Injectable AbstractLogRecordBolt.RecordCollector collector, @Mocked LogRecord logRecord) { String tupleValue = "tuple-value"; new Expectations(bolt) {{ input.getValue(0); result = tupleValue.getBytes(); new LogRecord(anyString); result = logRecord; bolt.parseJson(anyString, logRecord); collector.emit(logRecord); }}; bolt.execute(input, collector); }