@Test public void testConfigNormal() throws DocumentException { Configure configure = ConfigureReader.reader("src/test/resources/configure.xml"); Assert.assertEquals(configure.getOracleSid(), "100"); Assert.assertEquals(configure.getDatahubEndpoint(), "YOUR_DATAHUB_ENDPOINT"); Assert.assertEquals(configure.getDatahubAccessId(), "YOUR_DATAHUB_ACCESS_ID"); Assert.assertEquals(configure.getDatahubAccessKey(), "YOUR_DATAHUB_ACCESS_KEY"); Assert.assertTrue(configure.isEnablePb()); Assert.assertEquals(configure.getCompressType(), "LZ4"); Assert.assertEquals(configure.getBatchSize(), 1111); Assert.assertEquals(configure.getBatchTimeoutMs(), 5000); Assert.assertTrue(configure.isDirtyDataContinue()); Assert.assertEquals(configure.getDirtyDataFile(), "datahub_ogg_plugin.dirty.test"); Assert.assertEquals(configure.getDirtyDataFileMaxSize(), 200 * 1000000); Assert.assertEquals(configure.getRetryTimes(), 35); Assert.assertEquals(configure.getRetryIntervalMs(), 1000); Assert.assertEquals(configure.getCheckPointFileName(), "datahub_ogg_plugin.chk.test"); Assert.assertEquals(configure.getCharsetName(), "utf-8"); Assert.assertTrue(configure.isCheckPointFileDisable()); Assert.assertTrue(configure.isCommitFlush()); Assert.assertFalse(configure.isReportMetric()); Assert.assertEquals(configure.getReportMetricIntervalMs(), 5 * 60 * 1000); Map<String, TableMapping> tableMappingMap = configure.getTableMappings(); Assert.assertEquals(tableMappingMap.size(), 2); Assert.assertEquals(tableMappingMap.keySet(), Sets.newHashSet("ogg_test.test2", "t_schema.t_person")); TableMapping table1 = configure.getTableMapping("t_schema.t_person"); Assert.assertEquals(table1.getOracleSchema(), "t_schema"); Assert.assertEquals(table1.getOracleTableName(), "t_person"); Assert.assertEquals(table1.getOracleFullTableName(), "t_schema.t_person"); Assert.assertEquals(table1.getProjectName(), "t_project"); Assert.assertEquals(table1.getTopicName(), "ogg_test_normal"); Assert.assertEquals(table1.getAccessId(), "accessId"); Assert.assertEquals(table1.getAccessKey(), "accessKey"); Assert.assertEquals(table1.getShardIds(), Arrays.asList("0", "1")); Assert.assertNull(table1.getRecordSchema()); Assert.assertEquals(table1.getRowIdColumn(), "row_id"); Assert.assertEquals(table1.getcTypeColumn(), "ctype"); Assert.assertEquals(table1.getcTimeColumn(), "ctime"); Assert.assertEquals(table1.getcIdColumn(), "cid"); Map<String, String> map = table1.getConstColumnMappings(); Assert.assertEquals(map.size(), 2); Assert.assertEquals(map.get("const1"), "3"); Assert.assertEquals(map.get("const4"), "abcd"); Assert.assertEquals(table1.getcIdColumn(), "cid"); Map<String, ColumnMapping> columns = table1.getColumnMappings(); Assert.assertEquals(columns.size(), 2); Assert.assertEquals(columns.keySet(), Sets.newHashSet("c1", "c2")); ColumnMapping column1 = columns.get("c1"); Assert.assertEquals(column1.getDest(), "c1"); Assert.assertNull(column1.getDestOld()); Assert.assertFalse(column1.isShardColumn()); Assert.assertTrue(column1.isDateFormat()); Assert.assertFalse(column1.isKeyColumn()); Assert.assertTrue(column1.isDefaultCharset()); ColumnMapping column2 = columns.get("c2"); Assert.assertEquals(column2.getDest(), "c2"); Assert.assertEquals(column2.getDestOld(), "c2_old"); Assert.assertTrue(column2.isShardColumn()); Assert.assertFalse(column2.isDateFormat()); Assert.assertEquals(column2.getDateFormat(), "yyyy-MM-dd:HH:mm:ss"); Assert.assertFalse(column2.isKeyColumn()); Assert.assertFalse(column2.isDefaultCharset()); TableMapping table2 = configure.getTableMapping("ogg_test.test2"); Assert.assertEquals(table2.getOracleSchema(), "ogg_test"); Assert.assertEquals(table2.getOracleTableName(), "test2"); Assert.assertEquals(table2.getOracleFullTableName(), "ogg_test.test2"); Assert.assertEquals(table2.getProjectName(), "YOUR_DATAHUB_PROJECT"); Assert.assertEquals(table2.getTopicName(), "ogg_test_normal"); Assert.assertEquals(table2.getAccessId(), "YOUR_DATAHUB_ACCESS_ID"); Assert.assertEquals(table2.getAccessKey(), "YOUR_DATAHUB_ACCESS_KEY"); Assert.assertTrue(table2.getShardIds().isEmpty()); Assert.assertNull(table2.getRecordSchema()); Assert.assertNull(table2.getRowIdColumn()); Assert.assertNull(table2.getcTypeColumn()); Assert.assertNull(table2.getcTimeColumn()); Assert.assertNull(table2.getcIdColumn()); Assert.assertTrue(table2.getConstColumnMappings().isEmpty()); columns = table2.getColumnMappings(); Assert.assertEquals(columns.size(), 2); column1 = columns.get("c1"); Assert.assertEquals(column1.getDest(), "c1"); Assert.assertEquals(column1.getDateFormat(), "yyyy-MM-dd"); Assert.assertNull(column1.getDestOld()); column2 = columns.get("c2"); Assert.assertNull(column2.getDateFormat()); Assert.assertNull(column2.getSimpleDateFormat()); }
public static Configure reader(String configueFileName) throws DocumentException { logger.info("Begin read configure[{}]", configueFileName); Configure configure = new Configure(); SAXReader reader = new SAXReader(); File file = new File(configueFileName); Document document = reader.read(file); Element root = document.getRootElement(); String elementText = root.elementTextTrim("batchSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("batchTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("dirtyDataContinue"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataContinue(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("dirtyDataFile"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFile(elementText); } elementText = root.elementTextTrim("dirtyDataFileMaxSize"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFileMaxSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryTimes"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryTimes(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryInterval"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryIntervalMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("disableCheckPointFile"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileDisable(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("checkPointFileName"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileName(elementText); } elementText = root.elementTextTrim("charset"); if (StringUtils.isNotBlank(elementText)) { configure.setCharsetName(elementText); } elementText = root.elementTextTrim("buildRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("buildRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("commitFlush"); if (StringUtils.isNotBlank(elementText)) { configure.setCommitFlush(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetric"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetric(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetricIntervalMs"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetricIntervalMs(Integer.parseInt(elementText)); } Element element = root.element("defaultOracleConfigure"); if (element == null) { throw new RuntimeException("defaultOracleConfigure is null"); } elementText = element.elementTextTrim("sid"); if (StringUtils.isNotBlank(elementText)){ configure.setOracleSid(elementText); } String defaultOracleSchema = element.elementTextTrim("schema"); element = root.element("defaultDatahubConfigure"); if (element == null) { throw new RuntimeException("defaultDatahubConfigure is null"); } String endPoint = element.elementText("endPoint"); if (StringUtils.isBlank(endPoint)) { throw new RuntimeException("defaultDatahubConfigure.endPoint is null"); } configure.setDatahubEndpoint(endPoint); String defaultDatahubAccessID = element.elementText("accessId"); if (StringUtils.isNotBlank(defaultDatahubAccessID)) { configure.setDatahubAccessId(defaultDatahubAccessID); } String defaultDatahubAccessKey = element.elementText("accessKey"); if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { configure.setDatahubAccessKey(defaultDatahubAccessKey); } String defaultDatahubProject = element.elementText("project"); String defaultCTypeColumn = element.elementText("ctypeColumn"); String defaultCTimeColumn = element.elementText("ctimeColumn"); String defaultCidColumn = element.elementText("cidColumn"); String defaultConstColumnMapStr = element.elementText("constColumnMap"); Map<String, String> defalutConstColumnMappings = Maps.newHashMap(); parseConstColumnMap(defaultConstColumnMapStr, defalutConstColumnMappings); String compressType = element.elementText("compressType"); if (StringUtils.isNotBlank(compressType)) { configure.setCompressType(compressType); } String enablePb = element.elementText("enablePb"); if (StringUtils.isNotBlank(enablePb)) { configure.setEnablePb(Boolean.parseBoolean(enablePb)); } element = root.element("mappings"); if (element == null) { throw new RuntimeException("mappings is null"); } List<Element> mappingElements = element.elements("mapping"); System.out.println(mappingElements.size()); if (mappingElements == null || mappingElements.size() == 0) { throw new RuntimeException("mappings.mapping is null"); } for (Element e : mappingElements) { String oracleSchema = e.elementTextTrim("oracleSchema"); if (StringUtils.isNotBlank(oracleSchema)) { } else if (StringUtils.isNotBlank(defaultOracleSchema)) { oracleSchema = defaultOracleSchema; } else { throw new RuntimeException( "both mappings.mapping.oracleSchema and defaultOracleConfigure.schema is null"); } String oracleTable = e.elementTextTrim("oracleTable"); if (StringUtils.isBlank(oracleTable)) { throw new RuntimeException("mappings.mapping.oracleTable is null"); } String datahubProject = e.elementTextTrim("datahubProject"); if (StringUtils.isNotBlank(datahubProject)) { } else if (StringUtils.isNotBlank(defaultDatahubProject)) { datahubProject = defaultDatahubProject; } else { throw new RuntimeException( "both mappings.mapping.datahubProject and defaultDatahubConfigure.project is null"); } String datahubAccessId = e.elementTextTrim("datahubAccessId"); if (StringUtils.isNotBlank(datahubAccessId)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessID)) { datahubAccessId = defaultDatahubAccessID; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessId and defaultDatahubConfigure.accessId is null"); } String datahubAccessKey = e.elementTextTrim("datahubAccessKey"); if (StringUtils.isNotBlank(datahubAccessKey)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { datahubAccessKey = defaultDatahubAccessKey; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessKey and defaultDatahubConfigure.accessKey is null"); } String topicName = e.elementTextTrim("datahubTopic"); if (StringUtils.isBlank(topicName)) { throw new RuntimeException("mappings.mapping.datahubTopic is null"); } int buildSpeed = 0; elementText = e.elementText("buildSpeed"); if (StringUtils.isNotBlank(elementText)) { buildSpeed = Integer.parseInt(elementText); } String rowIdColumn = e.elementText("rowIdColumn"); String cTypeColumn = e.elementText("ctypeColumn"); cTypeColumn = StringUtils.isNotBlank(cTypeColumn) ? cTypeColumn : defaultCTypeColumn; String cTimeColumn = e.elementText("ctimeColumn"); cTimeColumn = StringUtils.isNotBlank(cTimeColumn) ? cTimeColumn : defaultCTimeColumn; String cIdColumn = e.elementText("cidColumn"); cIdColumn = StringUtils.isNotBlank(cIdColumn) ? cIdColumn : defaultCidColumn; String constColumnMapStr = e.elementText("constColumnMap"); Map<String, String> constColumnMappings = Maps.newHashMap(); parseConstColumnMap(constColumnMapStr, constColumnMappings); constColumnMappings = constColumnMappings.isEmpty() ? defalutConstColumnMappings : constColumnMappings; elementText = e.elementTextTrim("shardId"); List<String> shardIds = new ArrayList<String>(); paraseShardList(elementText, shardIds); TableMapping tableMapping = new TableMapping(); tableMapping.setOracleSchema(oracleSchema.toLowerCase()); tableMapping.setOracleTableName(oracleTable.toLowerCase()); tableMapping.setOracleFullTableName( tableMapping.getOracleSchema() + "." + tableMapping.getOracleTableName()); tableMapping.setProjectName(datahubProject); tableMapping.setTopicName(topicName); tableMapping.setAccessId(datahubAccessId); tableMapping.setAccessKey(datahubAccessKey); tableMapping.setRowIdColumn(rowIdColumn); tableMapping.setcTypeColumn(cTypeColumn); tableMapping.setcTimeColumn(cTimeColumn); tableMapping.setcIdColumn(cIdColumn); tableMapping.setConstColumnMappings(constColumnMappings); tableMapping.setShardIds(shardIds); if (!shardIds.isEmpty()) { tableMapping.setSetShardId(true); } if (buildSpeed > 0) { tableMapping.setBuildSpeed(buildSpeed); } configure.addTableMapping(tableMapping); Map<String, ColumnMapping> columnMappings = Maps.newHashMap(); tableMapping.setColumnMappings(columnMappings); Element columnMappingElement = e.element("columnMapping"); List<Element> columns = columnMappingElement.elements("column"); for (Element columnElement : columns) { String oracleColumnName = columnElement.attributeValue("src"); if (StringUtils.isBlank(oracleColumnName)) { throw new RuntimeException("Topic[" + topicName + "] src attribute is null"); } oracleColumnName = oracleColumnName.toLowerCase(); ColumnMapping columnMapping = new ColumnMapping(); columnMappings.put(oracleColumnName, columnMapping); columnMapping.setSrc(oracleColumnName); String dest = columnElement.attributeValue("dest"); if (StringUtils.isBlank(dest)) { throw new RuntimeException("Topic[" + topicName + "] dest attribute is null"); } columnMapping.setDest(dest); String destOld = columnElement.attributeValue("destOld"); if (StringUtils.isNotBlank(destOld)) { columnMapping.setDestOld(destOld); } String isShardColumn = columnElement.attributeValue("isShardColumn"); if (StringUtils.isNotBlank(isShardColumn) && Boolean.TRUE .equals(Boolean.valueOf(isShardColumn))) { tableMapping.setShardHash(true); columnMapping.setIsShardColumn(true); } else { columnMapping.setIsShardColumn(false); } String isKeyColumn = columnElement.attributeValue("isKeyColumn"); if (StringUtils.isNotBlank(isKeyColumn) && Boolean.TRUE .equals(Boolean.valueOf(isKeyColumn))) { columnMapping.setIsKeyColumn(true); } else { columnMapping.setIsKeyColumn(false); } String isDateFormat = columnElement.attributeValue("isDateFormat"); if (StringUtils.isNotBlank(isDateFormat)) { columnMapping.setIsDateFormat(Boolean.parseBoolean(isDateFormat)); } String dateFormat = columnElement.attributeValue("dateFormat"); if (StringUtils.isNotBlank(dateFormat)) { columnMapping.setDateFormat(dateFormat); } String charset = columnElement.attributeValue("isDefaultCharset"); if (StringUtils.isNotBlank(charset)) { columnMapping.setDefaultCharset(Boolean.parseBoolean(isDateFormat)); } } } logger.info("Read configure success: {} ", JsonHelper.beanToJson(configure)); return configure; }
ConfigureReader { public static Configure reader(String configueFileName) throws DocumentException { logger.info("Begin read configure[{}]", configueFileName); Configure configure = new Configure(); SAXReader reader = new SAXReader(); File file = new File(configueFileName); Document document = reader.read(file); Element root = document.getRootElement(); String elementText = root.elementTextTrim("batchSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("batchTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("dirtyDataContinue"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataContinue(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("dirtyDataFile"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFile(elementText); } elementText = root.elementTextTrim("dirtyDataFileMaxSize"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFileMaxSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryTimes"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryTimes(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryInterval"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryIntervalMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("disableCheckPointFile"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileDisable(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("checkPointFileName"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileName(elementText); } elementText = root.elementTextTrim("charset"); if (StringUtils.isNotBlank(elementText)) { configure.setCharsetName(elementText); } elementText = root.elementTextTrim("buildRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("buildRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("commitFlush"); if (StringUtils.isNotBlank(elementText)) { configure.setCommitFlush(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetric"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetric(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetricIntervalMs"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetricIntervalMs(Integer.parseInt(elementText)); } Element element = root.element("defaultOracleConfigure"); if (element == null) { throw new RuntimeException("defaultOracleConfigure is null"); } elementText = element.elementTextTrim("sid"); if (StringUtils.isNotBlank(elementText)){ configure.setOracleSid(elementText); } String defaultOracleSchema = element.elementTextTrim("schema"); element = root.element("defaultDatahubConfigure"); if (element == null) { throw new RuntimeException("defaultDatahubConfigure is null"); } String endPoint = element.elementText("endPoint"); if (StringUtils.isBlank(endPoint)) { throw new RuntimeException("defaultDatahubConfigure.endPoint is null"); } configure.setDatahubEndpoint(endPoint); String defaultDatahubAccessID = element.elementText("accessId"); if (StringUtils.isNotBlank(defaultDatahubAccessID)) { configure.setDatahubAccessId(defaultDatahubAccessID); } String defaultDatahubAccessKey = element.elementText("accessKey"); if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { configure.setDatahubAccessKey(defaultDatahubAccessKey); } String defaultDatahubProject = element.elementText("project"); String defaultCTypeColumn = element.elementText("ctypeColumn"); String defaultCTimeColumn = element.elementText("ctimeColumn"); String defaultCidColumn = element.elementText("cidColumn"); String defaultConstColumnMapStr = element.elementText("constColumnMap"); Map<String, String> defalutConstColumnMappings = Maps.newHashMap(); parseConstColumnMap(defaultConstColumnMapStr, defalutConstColumnMappings); String compressType = element.elementText("compressType"); if (StringUtils.isNotBlank(compressType)) { configure.setCompressType(compressType); } String enablePb = element.elementText("enablePb"); if (StringUtils.isNotBlank(enablePb)) { configure.setEnablePb(Boolean.parseBoolean(enablePb)); } element = root.element("mappings"); if (element == null) { throw new RuntimeException("mappings is null"); } List<Element> mappingElements = element.elements("mapping"); System.out.println(mappingElements.size()); if (mappingElements == null || mappingElements.size() == 0) { throw new RuntimeException("mappings.mapping is null"); } for (Element e : mappingElements) { String oracleSchema = e.elementTextTrim("oracleSchema"); if (StringUtils.isNotBlank(oracleSchema)) { } else if (StringUtils.isNotBlank(defaultOracleSchema)) { oracleSchema = defaultOracleSchema; } else { throw new RuntimeException( "both mappings.mapping.oracleSchema and defaultOracleConfigure.schema is null"); } String oracleTable = e.elementTextTrim("oracleTable"); if (StringUtils.isBlank(oracleTable)) { throw new RuntimeException("mappings.mapping.oracleTable is null"); } String datahubProject = e.elementTextTrim("datahubProject"); if (StringUtils.isNotBlank(datahubProject)) { } else if (StringUtils.isNotBlank(defaultDatahubProject)) { datahubProject = defaultDatahubProject; } else { throw new RuntimeException( "both mappings.mapping.datahubProject and defaultDatahubConfigure.project is null"); } String datahubAccessId = e.elementTextTrim("datahubAccessId"); if (StringUtils.isNotBlank(datahubAccessId)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessID)) { datahubAccessId = defaultDatahubAccessID; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessId and defaultDatahubConfigure.accessId is null"); } String datahubAccessKey = e.elementTextTrim("datahubAccessKey"); if (StringUtils.isNotBlank(datahubAccessKey)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { datahubAccessKey = defaultDatahubAccessKey; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessKey and defaultDatahubConfigure.accessKey is null"); } String topicName = e.elementTextTrim("datahubTopic"); if (StringUtils.isBlank(topicName)) { throw new RuntimeException("mappings.mapping.datahubTopic is null"); } int buildSpeed = 0; elementText = e.elementText("buildSpeed"); if (StringUtils.isNotBlank(elementText)) { buildSpeed = Integer.parseInt(elementText); } String rowIdColumn = e.elementText("rowIdColumn"); String cTypeColumn = e.elementText("ctypeColumn"); cTypeColumn = StringUtils.isNotBlank(cTypeColumn) ? cTypeColumn : defaultCTypeColumn; String cTimeColumn = e.elementText("ctimeColumn"); cTimeColumn = StringUtils.isNotBlank(cTimeColumn) ? cTimeColumn : defaultCTimeColumn; String cIdColumn = e.elementText("cidColumn"); cIdColumn = StringUtils.isNotBlank(cIdColumn) ? cIdColumn : defaultCidColumn; String constColumnMapStr = e.elementText("constColumnMap"); Map<String, String> constColumnMappings = Maps.newHashMap(); parseConstColumnMap(constColumnMapStr, constColumnMappings); constColumnMappings = constColumnMappings.isEmpty() ? defalutConstColumnMappings : constColumnMappings; elementText = e.elementTextTrim("shardId"); List<String> shardIds = new ArrayList<String>(); paraseShardList(elementText, shardIds); TableMapping tableMapping = new TableMapping(); tableMapping.setOracleSchema(oracleSchema.toLowerCase()); tableMapping.setOracleTableName(oracleTable.toLowerCase()); tableMapping.setOracleFullTableName( tableMapping.getOracleSchema() + "." + tableMapping.getOracleTableName()); tableMapping.setProjectName(datahubProject); tableMapping.setTopicName(topicName); tableMapping.setAccessId(datahubAccessId); tableMapping.setAccessKey(datahubAccessKey); tableMapping.setRowIdColumn(rowIdColumn); tableMapping.setcTypeColumn(cTypeColumn); tableMapping.setcTimeColumn(cTimeColumn); tableMapping.setcIdColumn(cIdColumn); tableMapping.setConstColumnMappings(constColumnMappings); tableMapping.setShardIds(shardIds); if (!shardIds.isEmpty()) { tableMapping.setSetShardId(true); } if (buildSpeed > 0) { tableMapping.setBuildSpeed(buildSpeed); } configure.addTableMapping(tableMapping); Map<String, ColumnMapping> columnMappings = Maps.newHashMap(); tableMapping.setColumnMappings(columnMappings); Element columnMappingElement = e.element("columnMapping"); List<Element> columns = columnMappingElement.elements("column"); for (Element columnElement : columns) { String oracleColumnName = columnElement.attributeValue("src"); if (StringUtils.isBlank(oracleColumnName)) { throw new RuntimeException("Topic[" + topicName + "] src attribute is null"); } oracleColumnName = oracleColumnName.toLowerCase(); ColumnMapping columnMapping = new ColumnMapping(); columnMappings.put(oracleColumnName, columnMapping); columnMapping.setSrc(oracleColumnName); String dest = columnElement.attributeValue("dest"); if (StringUtils.isBlank(dest)) { throw new RuntimeException("Topic[" + topicName + "] dest attribute is null"); } columnMapping.setDest(dest); String destOld = columnElement.attributeValue("destOld"); if (StringUtils.isNotBlank(destOld)) { columnMapping.setDestOld(destOld); } String isShardColumn = columnElement.attributeValue("isShardColumn"); if (StringUtils.isNotBlank(isShardColumn) && Boolean.TRUE .equals(Boolean.valueOf(isShardColumn))) { tableMapping.setShardHash(true); columnMapping.setIsShardColumn(true); } else { columnMapping.setIsShardColumn(false); } String isKeyColumn = columnElement.attributeValue("isKeyColumn"); if (StringUtils.isNotBlank(isKeyColumn) && Boolean.TRUE .equals(Boolean.valueOf(isKeyColumn))) { columnMapping.setIsKeyColumn(true); } else { columnMapping.setIsKeyColumn(false); } String isDateFormat = columnElement.attributeValue("isDateFormat"); if (StringUtils.isNotBlank(isDateFormat)) { columnMapping.setIsDateFormat(Boolean.parseBoolean(isDateFormat)); } String dateFormat = columnElement.attributeValue("dateFormat"); if (StringUtils.isNotBlank(dateFormat)) { columnMapping.setDateFormat(dateFormat); } String charset = columnElement.attributeValue("isDefaultCharset"); if (StringUtils.isNotBlank(charset)) { columnMapping.setDefaultCharset(Boolean.parseBoolean(isDateFormat)); } } } logger.info("Read configure success: {} ", JsonHelper.beanToJson(configure)); return configure; } }
ConfigureReader { public static Configure reader(String configueFileName) throws DocumentException { logger.info("Begin read configure[{}]", configueFileName); Configure configure = new Configure(); SAXReader reader = new SAXReader(); File file = new File(configueFileName); Document document = reader.read(file); Element root = document.getRootElement(); String elementText = root.elementTextTrim("batchSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("batchTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("dirtyDataContinue"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataContinue(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("dirtyDataFile"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFile(elementText); } elementText = root.elementTextTrim("dirtyDataFileMaxSize"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFileMaxSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryTimes"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryTimes(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryInterval"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryIntervalMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("disableCheckPointFile"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileDisable(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("checkPointFileName"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileName(elementText); } elementText = root.elementTextTrim("charset"); if (StringUtils.isNotBlank(elementText)) { configure.setCharsetName(elementText); } elementText = root.elementTextTrim("buildRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("buildRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("commitFlush"); if (StringUtils.isNotBlank(elementText)) { configure.setCommitFlush(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetric"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetric(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetricIntervalMs"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetricIntervalMs(Integer.parseInt(elementText)); } Element element = root.element("defaultOracleConfigure"); if (element == null) { throw new RuntimeException("defaultOracleConfigure is null"); } elementText = element.elementTextTrim("sid"); if (StringUtils.isNotBlank(elementText)){ configure.setOracleSid(elementText); } String defaultOracleSchema = element.elementTextTrim("schema"); element = root.element("defaultDatahubConfigure"); if (element == null) { throw new RuntimeException("defaultDatahubConfigure is null"); } String endPoint = element.elementText("endPoint"); if (StringUtils.isBlank(endPoint)) { throw new RuntimeException("defaultDatahubConfigure.endPoint is null"); } configure.setDatahubEndpoint(endPoint); String defaultDatahubAccessID = element.elementText("accessId"); if (StringUtils.isNotBlank(defaultDatahubAccessID)) { configure.setDatahubAccessId(defaultDatahubAccessID); } String defaultDatahubAccessKey = element.elementText("accessKey"); if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { configure.setDatahubAccessKey(defaultDatahubAccessKey); } String defaultDatahubProject = element.elementText("project"); String defaultCTypeColumn = element.elementText("ctypeColumn"); String defaultCTimeColumn = element.elementText("ctimeColumn"); String defaultCidColumn = element.elementText("cidColumn"); String defaultConstColumnMapStr = element.elementText("constColumnMap"); Map<String, String> defalutConstColumnMappings = Maps.newHashMap(); parseConstColumnMap(defaultConstColumnMapStr, defalutConstColumnMappings); String compressType = element.elementText("compressType"); if (StringUtils.isNotBlank(compressType)) { configure.setCompressType(compressType); } String enablePb = element.elementText("enablePb"); if (StringUtils.isNotBlank(enablePb)) { configure.setEnablePb(Boolean.parseBoolean(enablePb)); } element = root.element("mappings"); if (element == null) { throw new RuntimeException("mappings is null"); } List<Element> mappingElements = element.elements("mapping"); System.out.println(mappingElements.size()); if (mappingElements == null || mappingElements.size() == 0) { throw new RuntimeException("mappings.mapping is null"); } for (Element e : mappingElements) { String oracleSchema = e.elementTextTrim("oracleSchema"); if (StringUtils.isNotBlank(oracleSchema)) { } else if (StringUtils.isNotBlank(defaultOracleSchema)) { oracleSchema = defaultOracleSchema; } else { throw new RuntimeException( "both mappings.mapping.oracleSchema and defaultOracleConfigure.schema is null"); } String oracleTable = e.elementTextTrim("oracleTable"); if (StringUtils.isBlank(oracleTable)) { throw new RuntimeException("mappings.mapping.oracleTable is null"); } String datahubProject = e.elementTextTrim("datahubProject"); if (StringUtils.isNotBlank(datahubProject)) { } else if (StringUtils.isNotBlank(defaultDatahubProject)) { datahubProject = defaultDatahubProject; } else { throw new RuntimeException( "both mappings.mapping.datahubProject and defaultDatahubConfigure.project is null"); } String datahubAccessId = e.elementTextTrim("datahubAccessId"); if (StringUtils.isNotBlank(datahubAccessId)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessID)) { datahubAccessId = defaultDatahubAccessID; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessId and defaultDatahubConfigure.accessId is null"); } String datahubAccessKey = e.elementTextTrim("datahubAccessKey"); if (StringUtils.isNotBlank(datahubAccessKey)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { datahubAccessKey = defaultDatahubAccessKey; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessKey and defaultDatahubConfigure.accessKey is null"); } String topicName = e.elementTextTrim("datahubTopic"); if (StringUtils.isBlank(topicName)) { throw new RuntimeException("mappings.mapping.datahubTopic is null"); } int buildSpeed = 0; elementText = e.elementText("buildSpeed"); if (StringUtils.isNotBlank(elementText)) { buildSpeed = Integer.parseInt(elementText); } String rowIdColumn = e.elementText("rowIdColumn"); String cTypeColumn = e.elementText("ctypeColumn"); cTypeColumn = StringUtils.isNotBlank(cTypeColumn) ? cTypeColumn : defaultCTypeColumn; String cTimeColumn = e.elementText("ctimeColumn"); cTimeColumn = StringUtils.isNotBlank(cTimeColumn) ? cTimeColumn : defaultCTimeColumn; String cIdColumn = e.elementText("cidColumn"); cIdColumn = StringUtils.isNotBlank(cIdColumn) ? cIdColumn : defaultCidColumn; String constColumnMapStr = e.elementText("constColumnMap"); Map<String, String> constColumnMappings = Maps.newHashMap(); parseConstColumnMap(constColumnMapStr, constColumnMappings); constColumnMappings = constColumnMappings.isEmpty() ? defalutConstColumnMappings : constColumnMappings; elementText = e.elementTextTrim("shardId"); List<String> shardIds = new ArrayList<String>(); paraseShardList(elementText, shardIds); TableMapping tableMapping = new TableMapping(); tableMapping.setOracleSchema(oracleSchema.toLowerCase()); tableMapping.setOracleTableName(oracleTable.toLowerCase()); tableMapping.setOracleFullTableName( tableMapping.getOracleSchema() + "." + tableMapping.getOracleTableName()); tableMapping.setProjectName(datahubProject); tableMapping.setTopicName(topicName); tableMapping.setAccessId(datahubAccessId); tableMapping.setAccessKey(datahubAccessKey); tableMapping.setRowIdColumn(rowIdColumn); tableMapping.setcTypeColumn(cTypeColumn); tableMapping.setcTimeColumn(cTimeColumn); tableMapping.setcIdColumn(cIdColumn); tableMapping.setConstColumnMappings(constColumnMappings); tableMapping.setShardIds(shardIds); if (!shardIds.isEmpty()) { tableMapping.setSetShardId(true); } if (buildSpeed > 0) { tableMapping.setBuildSpeed(buildSpeed); } configure.addTableMapping(tableMapping); Map<String, ColumnMapping> columnMappings = Maps.newHashMap(); tableMapping.setColumnMappings(columnMappings); Element columnMappingElement = e.element("columnMapping"); List<Element> columns = columnMappingElement.elements("column"); for (Element columnElement : columns) { String oracleColumnName = columnElement.attributeValue("src"); if (StringUtils.isBlank(oracleColumnName)) { throw new RuntimeException("Topic[" + topicName + "] src attribute is null"); } oracleColumnName = oracleColumnName.toLowerCase(); ColumnMapping columnMapping = new ColumnMapping(); columnMappings.put(oracleColumnName, columnMapping); columnMapping.setSrc(oracleColumnName); String dest = columnElement.attributeValue("dest"); if (StringUtils.isBlank(dest)) { throw new RuntimeException("Topic[" + topicName + "] dest attribute is null"); } columnMapping.setDest(dest); String destOld = columnElement.attributeValue("destOld"); if (StringUtils.isNotBlank(destOld)) { columnMapping.setDestOld(destOld); } String isShardColumn = columnElement.attributeValue("isShardColumn"); if (StringUtils.isNotBlank(isShardColumn) && Boolean.TRUE .equals(Boolean.valueOf(isShardColumn))) { tableMapping.setShardHash(true); columnMapping.setIsShardColumn(true); } else { columnMapping.setIsShardColumn(false); } String isKeyColumn = columnElement.attributeValue("isKeyColumn"); if (StringUtils.isNotBlank(isKeyColumn) && Boolean.TRUE .equals(Boolean.valueOf(isKeyColumn))) { columnMapping.setIsKeyColumn(true); } else { columnMapping.setIsKeyColumn(false); } String isDateFormat = columnElement.attributeValue("isDateFormat"); if (StringUtils.isNotBlank(isDateFormat)) { columnMapping.setIsDateFormat(Boolean.parseBoolean(isDateFormat)); } String dateFormat = columnElement.attributeValue("dateFormat"); if (StringUtils.isNotBlank(dateFormat)) { columnMapping.setDateFormat(dateFormat); } String charset = columnElement.attributeValue("isDefaultCharset"); if (StringUtils.isNotBlank(charset)) { columnMapping.setDefaultCharset(Boolean.parseBoolean(isDateFormat)); } } } logger.info("Read configure success: {} ", JsonHelper.beanToJson(configure)); return configure; }  }
ConfigureReader { public static Configure reader(String configueFileName) throws DocumentException { logger.info("Begin read configure[{}]", configueFileName); Configure configure = new Configure(); SAXReader reader = new SAXReader(); File file = new File(configueFileName); Document document = reader.read(file); Element root = document.getRootElement(); String elementText = root.elementTextTrim("batchSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("batchTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("dirtyDataContinue"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataContinue(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("dirtyDataFile"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFile(elementText); } elementText = root.elementTextTrim("dirtyDataFileMaxSize"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFileMaxSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryTimes"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryTimes(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryInterval"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryIntervalMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("disableCheckPointFile"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileDisable(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("checkPointFileName"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileName(elementText); } elementText = root.elementTextTrim("charset"); if (StringUtils.isNotBlank(elementText)) { configure.setCharsetName(elementText); } elementText = root.elementTextTrim("buildRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("buildRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("commitFlush"); if (StringUtils.isNotBlank(elementText)) { configure.setCommitFlush(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetric"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetric(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetricIntervalMs"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetricIntervalMs(Integer.parseInt(elementText)); } Element element = root.element("defaultOracleConfigure"); if (element == null) { throw new RuntimeException("defaultOracleConfigure is null"); } elementText = element.elementTextTrim("sid"); if (StringUtils.isNotBlank(elementText)){ configure.setOracleSid(elementText); } String defaultOracleSchema = element.elementTextTrim("schema"); element = root.element("defaultDatahubConfigure"); if (element == null) { throw new RuntimeException("defaultDatahubConfigure is null"); } String endPoint = element.elementText("endPoint"); if (StringUtils.isBlank(endPoint)) { throw new RuntimeException("defaultDatahubConfigure.endPoint is null"); } configure.setDatahubEndpoint(endPoint); String defaultDatahubAccessID = element.elementText("accessId"); if (StringUtils.isNotBlank(defaultDatahubAccessID)) { configure.setDatahubAccessId(defaultDatahubAccessID); } String defaultDatahubAccessKey = element.elementText("accessKey"); if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { configure.setDatahubAccessKey(defaultDatahubAccessKey); } String defaultDatahubProject = element.elementText("project"); String defaultCTypeColumn = element.elementText("ctypeColumn"); String defaultCTimeColumn = element.elementText("ctimeColumn"); String defaultCidColumn = element.elementText("cidColumn"); String defaultConstColumnMapStr = element.elementText("constColumnMap"); Map<String, String> defalutConstColumnMappings = Maps.newHashMap(); parseConstColumnMap(defaultConstColumnMapStr, defalutConstColumnMappings); String compressType = element.elementText("compressType"); if (StringUtils.isNotBlank(compressType)) { configure.setCompressType(compressType); } String enablePb = element.elementText("enablePb"); if (StringUtils.isNotBlank(enablePb)) { configure.setEnablePb(Boolean.parseBoolean(enablePb)); } element = root.element("mappings"); if (element == null) { throw new RuntimeException("mappings is null"); } List<Element> mappingElements = element.elements("mapping"); System.out.println(mappingElements.size()); if (mappingElements == null || mappingElements.size() == 0) { throw new RuntimeException("mappings.mapping is null"); } for (Element e : mappingElements) { String oracleSchema = e.elementTextTrim("oracleSchema"); if (StringUtils.isNotBlank(oracleSchema)) { } else if (StringUtils.isNotBlank(defaultOracleSchema)) { oracleSchema = defaultOracleSchema; } else { throw new RuntimeException( "both mappings.mapping.oracleSchema and defaultOracleConfigure.schema is null"); } String oracleTable = e.elementTextTrim("oracleTable"); if (StringUtils.isBlank(oracleTable)) { throw new RuntimeException("mappings.mapping.oracleTable is null"); } String datahubProject = e.elementTextTrim("datahubProject"); if (StringUtils.isNotBlank(datahubProject)) { } else if (StringUtils.isNotBlank(defaultDatahubProject)) { datahubProject = defaultDatahubProject; } else { throw new RuntimeException( "both mappings.mapping.datahubProject and defaultDatahubConfigure.project is null"); } String datahubAccessId = e.elementTextTrim("datahubAccessId"); if (StringUtils.isNotBlank(datahubAccessId)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessID)) { datahubAccessId = defaultDatahubAccessID; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessId and defaultDatahubConfigure.accessId is null"); } String datahubAccessKey = e.elementTextTrim("datahubAccessKey"); if (StringUtils.isNotBlank(datahubAccessKey)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { datahubAccessKey = defaultDatahubAccessKey; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessKey and defaultDatahubConfigure.accessKey is null"); } String topicName = e.elementTextTrim("datahubTopic"); if (StringUtils.isBlank(topicName)) { throw new RuntimeException("mappings.mapping.datahubTopic is null"); } int buildSpeed = 0; elementText = e.elementText("buildSpeed"); if (StringUtils.isNotBlank(elementText)) { buildSpeed = Integer.parseInt(elementText); } String rowIdColumn = e.elementText("rowIdColumn"); String cTypeColumn = e.elementText("ctypeColumn"); cTypeColumn = StringUtils.isNotBlank(cTypeColumn) ? cTypeColumn : defaultCTypeColumn; String cTimeColumn = e.elementText("ctimeColumn"); cTimeColumn = StringUtils.isNotBlank(cTimeColumn) ? cTimeColumn : defaultCTimeColumn; String cIdColumn = e.elementText("cidColumn"); cIdColumn = StringUtils.isNotBlank(cIdColumn) ? cIdColumn : defaultCidColumn; String constColumnMapStr = e.elementText("constColumnMap"); Map<String, String> constColumnMappings = Maps.newHashMap(); parseConstColumnMap(constColumnMapStr, constColumnMappings); constColumnMappings = constColumnMappings.isEmpty() ? defalutConstColumnMappings : constColumnMappings; elementText = e.elementTextTrim("shardId"); List<String> shardIds = new ArrayList<String>(); paraseShardList(elementText, shardIds); TableMapping tableMapping = new TableMapping(); tableMapping.setOracleSchema(oracleSchema.toLowerCase()); tableMapping.setOracleTableName(oracleTable.toLowerCase()); tableMapping.setOracleFullTableName( tableMapping.getOracleSchema() + "." + tableMapping.getOracleTableName()); tableMapping.setProjectName(datahubProject); tableMapping.setTopicName(topicName); tableMapping.setAccessId(datahubAccessId); tableMapping.setAccessKey(datahubAccessKey); tableMapping.setRowIdColumn(rowIdColumn); tableMapping.setcTypeColumn(cTypeColumn); tableMapping.setcTimeColumn(cTimeColumn); tableMapping.setcIdColumn(cIdColumn); tableMapping.setConstColumnMappings(constColumnMappings); tableMapping.setShardIds(shardIds); if (!shardIds.isEmpty()) { tableMapping.setSetShardId(true); } if (buildSpeed > 0) { tableMapping.setBuildSpeed(buildSpeed); } configure.addTableMapping(tableMapping); Map<String, ColumnMapping> columnMappings = Maps.newHashMap(); tableMapping.setColumnMappings(columnMappings); Element columnMappingElement = e.element("columnMapping"); List<Element> columns = columnMappingElement.elements("column"); for (Element columnElement : columns) { String oracleColumnName = columnElement.attributeValue("src"); if (StringUtils.isBlank(oracleColumnName)) { throw new RuntimeException("Topic[" + topicName + "] src attribute is null"); } oracleColumnName = oracleColumnName.toLowerCase(); ColumnMapping columnMapping = new ColumnMapping(); columnMappings.put(oracleColumnName, columnMapping); columnMapping.setSrc(oracleColumnName); String dest = columnElement.attributeValue("dest"); if (StringUtils.isBlank(dest)) { throw new RuntimeException("Topic[" + topicName + "] dest attribute is null"); } columnMapping.setDest(dest); String destOld = columnElement.attributeValue("destOld"); if (StringUtils.isNotBlank(destOld)) { columnMapping.setDestOld(destOld); } String isShardColumn = columnElement.attributeValue("isShardColumn"); if (StringUtils.isNotBlank(isShardColumn) && Boolean.TRUE .equals(Boolean.valueOf(isShardColumn))) { tableMapping.setShardHash(true); columnMapping.setIsShardColumn(true); } else { columnMapping.setIsShardColumn(false); } String isKeyColumn = columnElement.attributeValue("isKeyColumn"); if (StringUtils.isNotBlank(isKeyColumn) && Boolean.TRUE .equals(Boolean.valueOf(isKeyColumn))) { columnMapping.setIsKeyColumn(true); } else { columnMapping.setIsKeyColumn(false); } String isDateFormat = columnElement.attributeValue("isDateFormat"); if (StringUtils.isNotBlank(isDateFormat)) { columnMapping.setIsDateFormat(Boolean.parseBoolean(isDateFormat)); } String dateFormat = columnElement.attributeValue("dateFormat"); if (StringUtils.isNotBlank(dateFormat)) { columnMapping.setDateFormat(dateFormat); } String charset = columnElement.attributeValue("isDefaultCharset"); if (StringUtils.isNotBlank(charset)) { columnMapping.setDefaultCharset(Boolean.parseBoolean(isDateFormat)); } } } logger.info("Read configure success: {} ", JsonHelper.beanToJson(configure)); return configure; }  static Configure reader(String configueFileName); }
ConfigureReader { public static Configure reader(String configueFileName) throws DocumentException { logger.info("Begin read configure[{}]", configueFileName); Configure configure = new Configure(); SAXReader reader = new SAXReader(); File file = new File(configueFileName); Document document = reader.read(file); Element root = document.getRootElement(); String elementText = root.elementTextTrim("batchSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("batchTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBatchTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("dirtyDataContinue"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataContinue(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("dirtyDataFile"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFile(elementText); } elementText = root.elementTextTrim("dirtyDataFileMaxSize"); if (StringUtils.isNotBlank(elementText)) { configure.setDirtyDataFileMaxSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryTimes"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryTimes(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("retryInterval"); if (StringUtils.isNotBlank(elementText)) { configure.setRetryIntervalMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("disableCheckPointFile"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileDisable(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("checkPointFileName"); if (StringUtils.isNotBlank(elementText)) { configure.setCheckPointFileName(elementText); } elementText = root.elementTextTrim("charset"); if (StringUtils.isNotBlank(elementText)) { configure.setCharsetName(elementText); } elementText = root.elementTextTrim("buildRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("buildRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setBuildRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueSize"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueSize(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("putRecordQueueTimeoutMs"); if (StringUtils.isNotBlank(elementText)) { configure.setPutRecordQueueTimeoutMs(Integer.parseInt(elementText)); } elementText = root.elementTextTrim("commitFlush"); if (StringUtils.isNotBlank(elementText)) { configure.setCommitFlush(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetric"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetric(Boolean.parseBoolean(elementText)); } elementText = root.elementTextTrim("reportMetricIntervalMs"); if (StringUtils.isNotBlank(elementText)) { configure.setReportMetricIntervalMs(Integer.parseInt(elementText)); } Element element = root.element("defaultOracleConfigure"); if (element == null) { throw new RuntimeException("defaultOracleConfigure is null"); } elementText = element.elementTextTrim("sid"); if (StringUtils.isNotBlank(elementText)){ configure.setOracleSid(elementText); } String defaultOracleSchema = element.elementTextTrim("schema"); element = root.element("defaultDatahubConfigure"); if (element == null) { throw new RuntimeException("defaultDatahubConfigure is null"); } String endPoint = element.elementText("endPoint"); if (StringUtils.isBlank(endPoint)) { throw new RuntimeException("defaultDatahubConfigure.endPoint is null"); } configure.setDatahubEndpoint(endPoint); String defaultDatahubAccessID = element.elementText("accessId"); if (StringUtils.isNotBlank(defaultDatahubAccessID)) { configure.setDatahubAccessId(defaultDatahubAccessID); } String defaultDatahubAccessKey = element.elementText("accessKey"); if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { configure.setDatahubAccessKey(defaultDatahubAccessKey); } String defaultDatahubProject = element.elementText("project"); String defaultCTypeColumn = element.elementText("ctypeColumn"); String defaultCTimeColumn = element.elementText("ctimeColumn"); String defaultCidColumn = element.elementText("cidColumn"); String defaultConstColumnMapStr = element.elementText("constColumnMap"); Map<String, String> defalutConstColumnMappings = Maps.newHashMap(); parseConstColumnMap(defaultConstColumnMapStr, defalutConstColumnMappings); String compressType = element.elementText("compressType"); if (StringUtils.isNotBlank(compressType)) { configure.setCompressType(compressType); } String enablePb = element.elementText("enablePb"); if (StringUtils.isNotBlank(enablePb)) { configure.setEnablePb(Boolean.parseBoolean(enablePb)); } element = root.element("mappings"); if (element == null) { throw new RuntimeException("mappings is null"); } List<Element> mappingElements = element.elements("mapping"); System.out.println(mappingElements.size()); if (mappingElements == null || mappingElements.size() == 0) { throw new RuntimeException("mappings.mapping is null"); } for (Element e : mappingElements) { String oracleSchema = e.elementTextTrim("oracleSchema"); if (StringUtils.isNotBlank(oracleSchema)) { } else if (StringUtils.isNotBlank(defaultOracleSchema)) { oracleSchema = defaultOracleSchema; } else { throw new RuntimeException( "both mappings.mapping.oracleSchema and defaultOracleConfigure.schema is null"); } String oracleTable = e.elementTextTrim("oracleTable"); if (StringUtils.isBlank(oracleTable)) { throw new RuntimeException("mappings.mapping.oracleTable is null"); } String datahubProject = e.elementTextTrim("datahubProject"); if (StringUtils.isNotBlank(datahubProject)) { } else if (StringUtils.isNotBlank(defaultDatahubProject)) { datahubProject = defaultDatahubProject; } else { throw new RuntimeException( "both mappings.mapping.datahubProject and defaultDatahubConfigure.project is null"); } String datahubAccessId = e.elementTextTrim("datahubAccessId"); if (StringUtils.isNotBlank(datahubAccessId)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessID)) { datahubAccessId = defaultDatahubAccessID; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessId and defaultDatahubConfigure.accessId is null"); } String datahubAccessKey = e.elementTextTrim("datahubAccessKey"); if (StringUtils.isNotBlank(datahubAccessKey)) { } else if (StringUtils.isNotBlank(defaultDatahubAccessKey)) { datahubAccessKey = defaultDatahubAccessKey; } else { throw new RuntimeException( "both mappings.mapping.datahubAccessKey and defaultDatahubConfigure.accessKey is null"); } String topicName = e.elementTextTrim("datahubTopic"); if (StringUtils.isBlank(topicName)) { throw new RuntimeException("mappings.mapping.datahubTopic is null"); } int buildSpeed = 0; elementText = e.elementText("buildSpeed"); if (StringUtils.isNotBlank(elementText)) { buildSpeed = Integer.parseInt(elementText); } String rowIdColumn = e.elementText("rowIdColumn"); String cTypeColumn = e.elementText("ctypeColumn"); cTypeColumn = StringUtils.isNotBlank(cTypeColumn) ? cTypeColumn : defaultCTypeColumn; String cTimeColumn = e.elementText("ctimeColumn"); cTimeColumn = StringUtils.isNotBlank(cTimeColumn) ? cTimeColumn : defaultCTimeColumn; String cIdColumn = e.elementText("cidColumn"); cIdColumn = StringUtils.isNotBlank(cIdColumn) ? cIdColumn : defaultCidColumn; String constColumnMapStr = e.elementText("constColumnMap"); Map<String, String> constColumnMappings = Maps.newHashMap(); parseConstColumnMap(constColumnMapStr, constColumnMappings); constColumnMappings = constColumnMappings.isEmpty() ? defalutConstColumnMappings : constColumnMappings; elementText = e.elementTextTrim("shardId"); List<String> shardIds = new ArrayList<String>(); paraseShardList(elementText, shardIds); TableMapping tableMapping = new TableMapping(); tableMapping.setOracleSchema(oracleSchema.toLowerCase()); tableMapping.setOracleTableName(oracleTable.toLowerCase()); tableMapping.setOracleFullTableName( tableMapping.getOracleSchema() + "." + tableMapping.getOracleTableName()); tableMapping.setProjectName(datahubProject); tableMapping.setTopicName(topicName); tableMapping.setAccessId(datahubAccessId); tableMapping.setAccessKey(datahubAccessKey); tableMapping.setRowIdColumn(rowIdColumn); tableMapping.setcTypeColumn(cTypeColumn); tableMapping.setcTimeColumn(cTimeColumn); tableMapping.setcIdColumn(cIdColumn); tableMapping.setConstColumnMappings(constColumnMappings); tableMapping.setShardIds(shardIds); if (!shardIds.isEmpty()) { tableMapping.setSetShardId(true); } if (buildSpeed > 0) { tableMapping.setBuildSpeed(buildSpeed); } configure.addTableMapping(tableMapping); Map<String, ColumnMapping> columnMappings = Maps.newHashMap(); tableMapping.setColumnMappings(columnMappings); Element columnMappingElement = e.element("columnMapping"); List<Element> columns = columnMappingElement.elements("column"); for (Element columnElement : columns) { String oracleColumnName = columnElement.attributeValue("src"); if (StringUtils.isBlank(oracleColumnName)) { throw new RuntimeException("Topic[" + topicName + "] src attribute is null"); } oracleColumnName = oracleColumnName.toLowerCase(); ColumnMapping columnMapping = new ColumnMapping(); columnMappings.put(oracleColumnName, columnMapping); columnMapping.setSrc(oracleColumnName); String dest = columnElement.attributeValue("dest"); if (StringUtils.isBlank(dest)) { throw new RuntimeException("Topic[" + topicName + "] dest attribute is null"); } columnMapping.setDest(dest); String destOld = columnElement.attributeValue("destOld"); if (StringUtils.isNotBlank(destOld)) { columnMapping.setDestOld(destOld); } String isShardColumn = columnElement.attributeValue("isShardColumn"); if (StringUtils.isNotBlank(isShardColumn) && Boolean.TRUE .equals(Boolean.valueOf(isShardColumn))) { tableMapping.setShardHash(true); columnMapping.setIsShardColumn(true); } else { columnMapping.setIsShardColumn(false); } String isKeyColumn = columnElement.attributeValue("isKeyColumn"); if (StringUtils.isNotBlank(isKeyColumn) && Boolean.TRUE .equals(Boolean.valueOf(isKeyColumn))) { columnMapping.setIsKeyColumn(true); } else { columnMapping.setIsKeyColumn(false); } String isDateFormat = columnElement.attributeValue("isDateFormat"); if (StringUtils.isNotBlank(isDateFormat)) { columnMapping.setIsDateFormat(Boolean.parseBoolean(isDateFormat)); } String dateFormat = columnElement.attributeValue("dateFormat"); if (StringUtils.isNotBlank(dateFormat)) { columnMapping.setDateFormat(dateFormat); } String charset = columnElement.attributeValue("isDefaultCharset"); if (StringUtils.isNotBlank(charset)) { columnMapping.setDefaultCharset(Boolean.parseBoolean(isDateFormat)); } } } logger.info("Read configure success: {} ", JsonHelper.beanToJson(configure)); return configure; }  static Configure reader(String configueFileName);  }