GarmadonMessage { public Map<String, Object> toMap(boolean includeDefaultValue, boolean addTypeMarker) { Map<String, Object> eventMap = ProtoConcatenator.concatToMap(timestamp, Arrays.asList(header, body), includeDefaultValue); if (addTypeMarker) { eventMap.put("event_type", GarmadonSerialization.getTypeName(type)); } if (GarmadonSerialization.TypeMarker.FS_EVENT == type) { String uri = (String) eventMap.get("uri"); eventMap.computeIfPresent("src_path", (k, v) -> ((String) v).replace(uri, "")); eventMap.computeIfPresent("dst_path", (k, v) -> ((String) v).replace(uri, "")); eventMap.computeIfPresent("uri", (k, v) -> UriHelper.getUniformizedUri((String) v)); } return eventMap; } GarmadonMessage(int type, long timestamp, EventHeaderProtos.Header header, Message body, CommittableOffset committableOffset); int getType(); long getTimestamp(); EventHeaderProtos.Header getHeader(); Message getBody(); CommittableOffset getCommittableOffset(); Map<String, Object> getHeaderMap(boolean includeDefaultValue); Map<String, Object> toMap(boolean includeDefaultValue, boolean addTypeMarker); Message toProto();  }
@Test public void testToMapNormalized() { long timestampMillis = System.currentTimeMillis(); EventHeaderProtos.Header header = EventHeaderProtos.Header.newBuilder().build(); DataAccessEventProtos.FsEvent.Builder fsEvent = DataAccessEventProtos.FsEvent.newBuilder() .setSrcPath("hdfs: .setDstPath("hdfs: .setAction("FS_RENAME") .setHdfsUser("mathieu") .setUri("hdfs: GarmadonMessage message = new GarmadonMessage(GarmadonSerialization.TypeMarker.FS_EVENT, timestampMillis, header, fsEvent.build(), null); Map<String, Object> map = message.toMap(false, false); HashMap<Object, Object> expected = new HashMap<>(); expected.put("src_path", "/src_file"); expected.put("dst_path", "/dst_file"); expected.put("uri", "hdfs: expected.put("hdfs_user", "mathieu"); expected.put("action", "FS_RENAME"); expected.put("timestamp", timestampMillis); TestCase.assertEquals(expected, map); }
@Test public void testToMapNormalizedWithPortInfo() { long timestampMillis = System.currentTimeMillis(); EventHeaderProtos.Header header = EventHeaderProtos.Header.newBuilder().build(); DataAccessEventProtos.FsEvent.Builder fsEvent = DataAccessEventProtos.FsEvent.newBuilder() .setSrcPath("hdfs: .setDstPath("hdfs: .setAction("FS_RENAME") .setHdfsUser("mathieu") .setUri("hdfs: GarmadonMessage message = new GarmadonMessage(GarmadonSerialization.TypeMarker.FS_EVENT, timestampMillis, header, fsEvent.build(), null); Map<String, Object> map = message.toMap(false, false); HashMap<Object, Object> expected = new HashMap<>(); expected.put("src_path", "/src_file"); expected.put("dst_path", "/dst_file"); expected.put("uri", "hdfs: expected.put("hdfs_user", "mathieu"); expected.put("action", "FS_RENAME"); expected.put("timestamp", timestampMillis); TestCase.assertEquals(expected, map); }
@Test public void testToMapOnInfinityDoubleReturnMaxNegativeValue() { long timestampMillis = System.currentTimeMillis(); EventHeaderProtos.Header header = EventHeaderProtos.Header.newBuilder().build(); FlinkEventProtos.OperatorEvent.Builder operatorEvent = FlinkEventProtos.OperatorEvent.newBuilder() .setRecordsLagMax(Double.NEGATIVE_INFINITY); GarmadonMessage message = new GarmadonMessage(GarmadonSerialization.TypeMarker.FLINK_OPERATOR_EVENT, timestampMillis, header, operatorEvent.build(), null); Map<String, Object> map = message.toMap(true, true); TestCase.assertEquals(-Double.MAX_VALUE, map.get("records_lag_max")); }