@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")); }
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 { 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 { 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); }
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(); }
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();  }