@Test(testName = "测试写入record") public void testWriteRecord() throws IOException { RecordSchema recordSchema = new RecordSchema() {{ addField(new Field("bigint_field", FieldType.BIGINT)); addField(new Field("timestamp_field", FieldType.TIMESTAMP)); addField(new Field("string_field", FieldType.STRING)); addField(new Field("double_field", FieldType.DOUBLE)); addField(new Field("boolean_field", FieldType.BOOLEAN)); addField(new Field("decimal_field", FieldType.DECIMAL)); }}; RecordEntry recordEntry = new RecordEntry(); TupleRecordData recordData = new TupleRecordData(recordSchema); long timestamp = System.currentTimeMillis(); recordData.setField("bigint_field", timestamp + 2); recordData.setField("timestamp_field", timestamp); recordData.setField("string_field", "abcdefg"); recordData.setField("double_field", 3.1415926); recordData.setField("boolean_field", true); recordData.setField("decimal_field", new BigDecimal("3.1415926")); recordEntry.setShardId("1"); recordEntry.setRecordData(recordData); BadOperateWriter.write(recordEntry, "ogg_test.t_person", "t_person", fileName, 10000, "record error"); String jsonStr = FileUtils.readFileToString(new File(fileName)); JSONObject object = JSON.parseObject(jsonStr); Assert.assertEquals(object.getString("oracleTable"), "ogg_test.t_person"); Assert.assertEquals(object.getString("topicName"), "t_person"); Assert.assertNull(object.getString("shardId"), null); Assert.assertEquals(object.getString("errorMessage"), "record error"); JSONObject record = object.getJSONObject("record"); Assert.assertEquals(record.getLongValue("bigint_field"), timestamp + 2); Assert.assertEquals(record.getLongValue("timestamp_field"), timestamp); Assert.assertEquals(record.getString("string_field"), "abcdefg"); Assert.assertEquals(record.getDoubleValue("double_field"), 3.1415926); Assert.assertTrue(record.getBooleanValue("boolean_field")); Assert.assertEquals(record.getBigDecimal("decimal_field"), new BigDecimal("3.1415926")); }
private static void write(Map<String, String> record, String oracleFullTableName, String topicName, String fileName, int maxFileSize, String msg) { checkFileSize(fileName, maxFileSize); DirtyRecordInfo dirtyRecordInfo = new DirtyRecordInfo(); dirtyRecordInfo.setOracleTable(oracleFullTableName); dirtyRecordInfo.setTopicName(topicName); dirtyRecordInfo.setShardId(null); dirtyRecordInfo.setErrorMessage(msg); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); dirtyRecordInfo.setErrorTime(simpleDateFormat.format(new Date())); dirtyRecordInfo.setRecord(record); try { BufferedWriter bw = new BufferedWriter(new FileWriter(fileName, true)); bw.write(JsonHelper.beanToJson(dirtyRecordInfo) + "\n"); bw.close(); } catch (IOException e) { logger.error("logBadOperation() failed. ", e); throw new RuntimeException("logBadOperation() failed. ", e); } }
BadOperateWriter { private static void write(Map<String, String> record, String oracleFullTableName, String topicName, String fileName, int maxFileSize, String msg) { checkFileSize(fileName, maxFileSize); DirtyRecordInfo dirtyRecordInfo = new DirtyRecordInfo(); dirtyRecordInfo.setOracleTable(oracleFullTableName); dirtyRecordInfo.setTopicName(topicName); dirtyRecordInfo.setShardId(null); dirtyRecordInfo.setErrorMessage(msg); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); dirtyRecordInfo.setErrorTime(simpleDateFormat.format(new Date())); dirtyRecordInfo.setRecord(record); try { BufferedWriter bw = new BufferedWriter(new FileWriter(fileName, true)); bw.write(JsonHelper.beanToJson(dirtyRecordInfo) + "\n"); bw.close(); } catch (IOException e) { logger.error("logBadOperation() failed. ", e); throw new RuntimeException("logBadOperation() failed. ", e); } } }
BadOperateWriter { private static void write(Map<String, String> record, String oracleFullTableName, String topicName, String fileName, int maxFileSize, String msg) { checkFileSize(fileName, maxFileSize); DirtyRecordInfo dirtyRecordInfo = new DirtyRecordInfo(); dirtyRecordInfo.setOracleTable(oracleFullTableName); dirtyRecordInfo.setTopicName(topicName); dirtyRecordInfo.setShardId(null); dirtyRecordInfo.setErrorMessage(msg); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); dirtyRecordInfo.setErrorTime(simpleDateFormat.format(new Date())); dirtyRecordInfo.setRecord(record); try { BufferedWriter bw = new BufferedWriter(new FileWriter(fileName, true)); bw.write(JsonHelper.beanToJson(dirtyRecordInfo) + "\n"); bw.close(); } catch (IOException e) { logger.error("logBadOperation() failed. ", e); throw new RuntimeException("logBadOperation() failed. ", e); } }  }
BadOperateWriter { private static void write(Map<String, String> record, String oracleFullTableName, String topicName, String fileName, int maxFileSize, String msg) { checkFileSize(fileName, maxFileSize); DirtyRecordInfo dirtyRecordInfo = new DirtyRecordInfo(); dirtyRecordInfo.setOracleTable(oracleFullTableName); dirtyRecordInfo.setTopicName(topicName); dirtyRecordInfo.setShardId(null); dirtyRecordInfo.setErrorMessage(msg); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); dirtyRecordInfo.setErrorTime(simpleDateFormat.format(new Date())); dirtyRecordInfo.setRecord(record); try { BufferedWriter bw = new BufferedWriter(new FileWriter(fileName, true)); bw.write(JsonHelper.beanToJson(dirtyRecordInfo) + "\n"); bw.close(); } catch (IOException e) { logger.error("logBadOperation() failed. ", e); throw new RuntimeException("logBadOperation() failed. ", e); } }  static void checkFileSize(String fileName, int maxFileSize); static void write(Op op, String oracleFullTableName, String topicName, String fileName,
                             int maxFileSize, String msg); static void write(RecordEntry recordEntry, String oracleFullTableName, String topicName,
                             String fileName, int maxFileSize, String msg); }
BadOperateWriter { private static void write(Map<String, String> record, String oracleFullTableName, String topicName, String fileName, int maxFileSize, String msg) { checkFileSize(fileName, maxFileSize); DirtyRecordInfo dirtyRecordInfo = new DirtyRecordInfo(); dirtyRecordInfo.setOracleTable(oracleFullTableName); dirtyRecordInfo.setTopicName(topicName); dirtyRecordInfo.setShardId(null); dirtyRecordInfo.setErrorMessage(msg); SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); dirtyRecordInfo.setErrorTime(simpleDateFormat.format(new Date())); dirtyRecordInfo.setRecord(record); try { BufferedWriter bw = new BufferedWriter(new FileWriter(fileName, true)); bw.write(JsonHelper.beanToJson(dirtyRecordInfo) + "\n"); bw.close(); } catch (IOException e) { logger.error("logBadOperation() failed. ", e); throw new RuntimeException("logBadOperation() failed. ", e); } }  static void checkFileSize(String fileName, int maxFileSize); static void write(Op op, String oracleFullTableName, String topicName, String fileName,
                             int maxFileSize, String msg); static void write(RecordEntry recordEntry, String oracleFullTableName, String topicName,
                             String fileName, int maxFileSize, String msg);  }