@Test public void testEmbeddedValueFaultTolerant() throws Exception { userNeedColNames = new String[] { "user_id", "user_sex" }; List<TblColRef> allCol = mockupTblColRefList(); TimedJsonStreamParser parser = new TimedJsonStreamParser(allCol, messageParserInfo()); Object msg = mapper.readValue(new File(jsonFilePath), mapType); ByteBuffer buffer = getJsonByteBuffer(msg); ConsumerRecord record = new ConsumerRecord<byte[], byte[]>("", 0, 0, null, buffer.array()); StreamingMessage parsedMsg = parser.parse(record); List<String> values = parsedMsg.getData(); assertEquals("4853763947", values.get(0)); assertEquals(StringUtils.EMPTY, values.get(1)); }
@Override public StreamingMessage parse(ConsumerRecord<byte[], byte[]> record) { try { Map<String, Object> message = mapper.readValue(parseToString(record.value()), mapType); root.clear(); root.putAll(message); String tsStr = root.get(tsColName).toString(); long t; if (StringUtils.isEmpty(tsStr)) { t = 0; } else { t = streamTimeParser.parseTime(tsStr); } ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { String columnName = column.getName(); TimeDerivedColumnType columnType = TimeDerivedColumnType.getTimeDerivedColumnType(columnName); if (columnType != null) { if (timeZoneOffset > 0 && TimeDerivedColumnType.isTimeDerivedColumnAboveDayLevel(columnName)) { result.add(String.valueOf(columnType.normalize(t + timeZoneOffset))); } else { result.add(String.valueOf(columnType.normalize(t))); } } else { Object value = root.get(columnName.toLowerCase(Locale.ROOT)); if (value == null) { String[] pathToValue = columnToSourceFieldMapping.get(columnName); if (pathToValue != null) { result.add(processMultiLevelJson(pathToValue, root)); } else { result.add(null); } } else { result.add(value.toString()); } } } return new StreamingMessage(result, new KafkaPartitionPosition(record.partition(), record.offset()), t, Collections.<String, Object> emptyMap()); } catch (IOException e) { logger.error("error", e); throw new RuntimeException(e); } }
TimedJsonStreamParser implements IStreamingMessageParser<ConsumerRecord<byte[], byte[]>> { @Override public StreamingMessage parse(ConsumerRecord<byte[], byte[]> record) { try { Map<String, Object> message = mapper.readValue(parseToString(record.value()), mapType); root.clear(); root.putAll(message); String tsStr = root.get(tsColName).toString(); long t; if (StringUtils.isEmpty(tsStr)) { t = 0; } else { t = streamTimeParser.parseTime(tsStr); } ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { String columnName = column.getName(); TimeDerivedColumnType columnType = TimeDerivedColumnType.getTimeDerivedColumnType(columnName); if (columnType != null) { if (timeZoneOffset > 0 && TimeDerivedColumnType.isTimeDerivedColumnAboveDayLevel(columnName)) { result.add(String.valueOf(columnType.normalize(t + timeZoneOffset))); } else { result.add(String.valueOf(columnType.normalize(t))); } } else { Object value = root.get(columnName.toLowerCase(Locale.ROOT)); if (value == null) { String[] pathToValue = columnToSourceFieldMapping.get(columnName); if (pathToValue != null) { result.add(processMultiLevelJson(pathToValue, root)); } else { result.add(null); } } else { result.add(value.toString()); } } } return new StreamingMessage(result, new KafkaPartitionPosition(record.partition(), record.offset()), t, Collections.<String, Object> emptyMap()); } catch (IOException e) { logger.error("error", e); throw new RuntimeException(e); } } }
TimedJsonStreamParser implements IStreamingMessageParser<ConsumerRecord<byte[], byte[]>> { @Override public StreamingMessage parse(ConsumerRecord<byte[], byte[]> record) { try { Map<String, Object> message = mapper.readValue(parseToString(record.value()), mapType); root.clear(); root.putAll(message); String tsStr = root.get(tsColName).toString(); long t; if (StringUtils.isEmpty(tsStr)) { t = 0; } else { t = streamTimeParser.parseTime(tsStr); } ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { String columnName = column.getName(); TimeDerivedColumnType columnType = TimeDerivedColumnType.getTimeDerivedColumnType(columnName); if (columnType != null) { if (timeZoneOffset > 0 && TimeDerivedColumnType.isTimeDerivedColumnAboveDayLevel(columnName)) { result.add(String.valueOf(columnType.normalize(t + timeZoneOffset))); } else { result.add(String.valueOf(columnType.normalize(t))); } } else { Object value = root.get(columnName.toLowerCase(Locale.ROOT)); if (value == null) { String[] pathToValue = columnToSourceFieldMapping.get(columnName); if (pathToValue != null) { result.add(processMultiLevelJson(pathToValue, root)); } else { result.add(null); } } else { result.add(value.toString()); } } } return new StreamingMessage(result, new KafkaPartitionPosition(record.partition(), record.offset()), t, Collections.<String, Object> emptyMap()); } catch (IOException e) { logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(CubeDesc cubeDesc, MessageParserInfo parserInfo);  TimedJsonStreamParser(List<TblColRef> cols, MessageParserInfo parserInfo); }
TimedJsonStreamParser implements IStreamingMessageParser<ConsumerRecord<byte[], byte[]>> { @Override public StreamingMessage parse(ConsumerRecord<byte[], byte[]> record) { try { Map<String, Object> message = mapper.readValue(parseToString(record.value()), mapType); root.clear(); root.putAll(message); String tsStr = root.get(tsColName).toString(); long t; if (StringUtils.isEmpty(tsStr)) { t = 0; } else { t = streamTimeParser.parseTime(tsStr); } ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { String columnName = column.getName(); TimeDerivedColumnType columnType = TimeDerivedColumnType.getTimeDerivedColumnType(columnName); if (columnType != null) { if (timeZoneOffset > 0 && TimeDerivedColumnType.isTimeDerivedColumnAboveDayLevel(columnName)) { result.add(String.valueOf(columnType.normalize(t + timeZoneOffset))); } else { result.add(String.valueOf(columnType.normalize(t))); } } else { Object value = root.get(columnName.toLowerCase(Locale.ROOT)); if (value == null) { String[] pathToValue = columnToSourceFieldMapping.get(columnName); if (pathToValue != null) { result.add(processMultiLevelJson(pathToValue, root)); } else { result.add(null); } } else { result.add(value.toString()); } } } return new StreamingMessage(result, new KafkaPartitionPosition(record.partition(), record.offset()), t, Collections.<String, Object> emptyMap()); } catch (IOException e) { logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(CubeDesc cubeDesc, MessageParserInfo parserInfo);  TimedJsonStreamParser(List<TblColRef> cols, MessageParserInfo parserInfo); static String objToString(Object value); @Override StreamingMessage parse(ConsumerRecord<byte[], byte[]> record); }
TimedJsonStreamParser implements IStreamingMessageParser<ConsumerRecord<byte[], byte[]>> { @Override public StreamingMessage parse(ConsumerRecord<byte[], byte[]> record) { try { Map<String, Object> message = mapper.readValue(parseToString(record.value()), mapType); root.clear(); root.putAll(message); String tsStr = root.get(tsColName).toString(); long t; if (StringUtils.isEmpty(tsStr)) { t = 0; } else { t = streamTimeParser.parseTime(tsStr); } ArrayList<String> result = Lists.newArrayList(); for (TblColRef column : allColumns) { String columnName = column.getName(); TimeDerivedColumnType columnType = TimeDerivedColumnType.getTimeDerivedColumnType(columnName); if (columnType != null) { if (timeZoneOffset > 0 && TimeDerivedColumnType.isTimeDerivedColumnAboveDayLevel(columnName)) { result.add(String.valueOf(columnType.normalize(t + timeZoneOffset))); } else { result.add(String.valueOf(columnType.normalize(t))); } } else { Object value = root.get(columnName.toLowerCase(Locale.ROOT)); if (value == null) { String[] pathToValue = columnToSourceFieldMapping.get(columnName); if (pathToValue != null) { result.add(processMultiLevelJson(pathToValue, root)); } else { result.add(null); } } else { result.add(value.toString()); } } } return new StreamingMessage(result, new KafkaPartitionPosition(record.partition(), record.offset()), t, Collections.<String, Object> emptyMap()); } catch (IOException e) { logger.error("error", e); throw new RuntimeException(e); } } TimedJsonStreamParser(CubeDesc cubeDesc, MessageParserInfo parserInfo);  TimedJsonStreamParser(List<TblColRef> cols, MessageParserInfo parserInfo); static String objToString(Object value); @Override StreamingMessage parse(ConsumerRecord<byte[], byte[]> record);  }