BaseTask implements InitableTask, StreamTask, ClosableTask { protected GenericData.Record getCamusHeaders(GenericRecord eventHeaders, Schema headerSchema, long tsNow) { GenericRecordBuilder builder = new GenericRecordBuilder(headerSchema); if (headerSchema.getField("created") != null) { builder.set("created", tsNow); } copyCamusTimestamp(eventHeaders, headerSchema, builder); copyAvroField(eventHeaders, headerSchema, builder, "id"); return builder.build(); }  ErrorHandler getErrorHandler(); void handleExpectedError(IncomingMessageEnvelope envelope, Exception e); @Override void init(Config config, TaskContext context); @Override final void process(IncomingMessageEnvelope envelope, MessageCollector collector, TaskCoordinator coordinator); @Override final void close(); static void run(String jobName);  }
@Test public void getCamusHeaders() { MyTask myTask = new MyTask(); GenericRecord inHdr = getHeader(); GenericData.Record outHdr = myTask.getCamusHeaders(inHdr, getHeaderSchema(), 21L); assertEquals("testId", outHdr.get("id").toString()); assertEquals(99L, ((Long)outHdr.get("time")).longValue()); assertEquals(21L, ((Long)outHdr.get("created")).longValue()); }
@Test public void getCamusHeadersNewToOld() { MyTask myTask = new MyTask(); GenericRecord inHdr = getHeader(); GenericData.Record outHdr = myTask.getCamusHeaders(inHdr, getOldHeaderSchema(), 21L); assertEquals("testId", outHdr.get("id").toString()); assertEquals(99L, ((Long)outHdr.get("timestamp")).longValue()); assertEquals(21L, ((Long)outHdr.get("created")).longValue()); }
@Test public void getCamusHeadersOldToNew() { MyTask myTask = new MyTask(); GenericRecord inHdr = getOldHeader(); GenericData.Record outHdr = myTask.getCamusHeaders(inHdr, getHeaderSchema(), 21L); assertEquals("testId", outHdr.get("id").toString()); assertEquals(99L, ((Long)outHdr.get("time")).longValue()); assertEquals(21L, ((Long)outHdr.get("created")).longValue()); }