ActionRequestUtil { public static ActionRequestKey getIndexRequestFromCamusHeader(IndexedRecord header, Schema headerSchema) { ActionRequestKey.Builder builder = ActionRequestKey.newBuilder(); builder.setAction(Action.INDEX); if (header != null && headerSchema != null) { Schema.Field tsField = headerSchema.getField("time"); Schema.Field idField = headerSchema.getField("id"); if (tsField == null) { tsField = headerSchema.getField("timestamp"); } if (tsField != null) { Object ts = header.get(tsField.pos()); if (ts instanceof Long) { builder.setEventTsUnixMs((Long) ts); builder.setPartitionTsUnixMs((Long) ts); } } if (idField != null) { Object id = header.get(idField.pos()); if (id instanceof CharSequence) { builder.setId((CharSequence) id); } } } return builder.build(); }  static ActionRequestKey getIndexRequestKeyFromCamus(IndexedRecord msg); static ActionRequestKey getIndexRequestFromCamusHeader(IndexedRecord header, Schema headerSchema);  }
@Test public void testGetIndexRequestKeyFromCamusHeader() throws Exception { ActionRequestKey requestKey = ActionRequestUtil.getIndexRequestFromCamusHeader(getHeader(), getHeaderSchema()); assertEquals(requestKey.getAction(), Action.INDEX); assertEquals(99L, requestKey.getPartitionTsUnixMs().longValue()); assertEquals(99L, requestKey.getEventTsUnixMs().longValue()); assertEquals("testId", requestKey.getId().toString()); }
@Test public void testGetIndexRequestKeyFromOldCamusHeader() throws Exception { ActionRequestKey requestKey = ActionRequestUtil.getIndexRequestFromCamusHeader(getOldHeader(), getOldHeaderSchema()); assertEquals(requestKey.getAction(), Action.INDEX); assertEquals(99L, requestKey.getPartitionTsUnixMs().longValue()); assertEquals(99L, requestKey.getEventTsUnixMs().longValue()); assertEquals("testId", requestKey.getId().toString()); }