diff --git a/admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/connector/SimpleAdminConnector.java b/admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/connector/SimpleAdminConnector.java index d3677280..419ba5f4 100644 --- a/admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/connector/SimpleAdminConnector.java +++ b/admin/admin-web/src/main/java/com/alibaba/otter/canal/admin/connector/SimpleAdminConnector.java @@ -108,9 +108,7 @@ public class SimpleAdminConnector implements AdminConnector { } connected = true; - } catch (IOException e) { - throw new ServiceException(e); - } catch (NoSuchAlgorithmException e) { + } catch (IOException | NoSuchAlgorithmException e) { throw new ServiceException(e); } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/DefaultPropertyNamePatternsMatcher.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/DefaultPropertyNamePatternsMatcher.java index 9f7ffe2e..289eda97 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/DefaultPropertyNamePatternsMatcher.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/bind/DefaultPropertyNamePatternsMatcher.java @@ -70,8 +70,8 @@ class DefaultPropertyNamePatternsMatcher implements PropertyNamePatternsMatcher return false; } } - for (int i = 0; i < match.length; i++) { - if (match[i]) { + for (boolean b : match) { + if (b) { return true; } } diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java index 0ee6d855..18c5ebb8 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/config/common/YamlProcessor.java @@ -130,8 +130,7 @@ public abstract class YamlProcessor { if (logger.isDebugEnabled()) { logger.debug("Loading from YAML: " + resource); } - Reader reader = new UnicodeReader(resource.getInputStream()); - try { + try (Reader reader = new UnicodeReader(resource.getInputStream())) { for (Object object : yaml.loadAll(reader)) { if (object != null && process(asMap(object), callback)) { count++; @@ -142,10 +141,8 @@ public abstract class YamlProcessor { } if (logger.isDebugEnabled()) { logger.debug( - "Loaded " + count + " document" + (count > 1 ? "s" : "") + " from YAML resource: " + resource); + "Loaded " + count + " document" + (count > 1 ? "s" : "") + " from YAML resource: " + resource); } - } finally { - reader.close(); } } catch (IOException ex) { handleProcessError(resource, ex); diff --git a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java index f7238eeb..e405db55 100644 --- a/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java +++ b/client-adapter/common/src/main/java/com/alibaba/otter/canal/client/adapter/support/ExtensionLoader.java @@ -6,6 +6,7 @@ import java.io.InputStreamReader; import java.net.MalformedURLException; import java.net.URL; import java.net.URLClassLoader; +import java.nio.charset.StandardCharsets; import java.nio.file.Path; import java.nio.file.Paths; import java.util.Arrays; @@ -317,10 +318,8 @@ public class ExtensionLoader { while (urls.hasMoreElements()) { URL url = urls.nextElement(); try { - BufferedReader reader = null; - try { - reader = new BufferedReader(new InputStreamReader(url.openStream(), "utf-8")); - String line = null; + try (BufferedReader reader = new BufferedReader(new InputStreamReader(url.openStream(), StandardCharsets.UTF_8))) { + String line; while ((line = reader.readLine()) != null) { final int ci = line.indexOf('#'); if (ci >= 0) line = line.substring(0, ci); @@ -340,10 +339,10 @@ public class ExtensionLoader { // classLoader); if (!type.isAssignableFrom(clazz)) { throw new IllegalStateException( - "Error when load extension class(interface: " + type - + ", class line: " + clazz.getName() - + "), class " + clazz.getName() - + "is not subtype of interface."); + "Error when load extension class(interface: " + type + + ", class line: " + clazz.getName() + + "), class " + clazz.getName() + + "is not subtype of interface."); } else { try { clazz.getConstructor(type); @@ -361,10 +360,10 @@ public class ExtensionLoader { } else if (c != clazz) { cachedNames.remove(clazz); throw new IllegalStateException( - "Duplicate extension " + type.getName() + " name " - + n + " on " - + c.getName() + " and " - + clazz.getName()); + "Duplicate extension " + type.getName() + " name " + + n + " on " + + c.getName() + " and " + + clazz.getName()); } } } @@ -373,19 +372,15 @@ public class ExtensionLoader { } } catch (Throwable t) { IllegalStateException e = new IllegalStateException( - "Failed to load extension class(interface: " + type + ", class line: " - + line + ") in " + url - + ", cause: " - + t.getMessage(), - t); + "Failed to load extension class(interface: " + type + ", class line: " + + line + ") in " + url + + ", cause: " + + t.getMessage(), + t); exceptions.put(line, e); } } } // end of while read lines - } finally { - if (reader != null) { - reader.close(); - } } } catch (Throwable t) { logger.error("Exception when load extension class(interface: " + type + ", class file: " + url diff --git a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java index 279f1150..0e930910 100644 --- a/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java +++ b/client-adapter/escore/src/main/java/com/alibaba/otter/canal/client/adapter/es/core/support/ESSyncUtil.java @@ -63,172 +63,187 @@ public class ESSyncUtil { return val; } Object res = null; - if ("integer".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).intValue(); - } else { - res = Integer.parseInt(val.toString()); - } - } else if ("long".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).longValue(); - } else { - res = Long.parseLong(val.toString()); - } - } else if ("short".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).shortValue(); - } else { - res = Short.parseShort(val.toString()); - } - } else if ("byte".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).byteValue(); - } else { - res = Byte.parseByte(val.toString()); - } - } else if ("double".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).doubleValue(); - } else { - res = Double.parseDouble(val.toString()); - } - } else if ("float".equals(esType) || "half_float".equals(esType) || "scaled_float".equals(esType)) { - if (val instanceof Number) { - res = ((Number) val).floatValue(); - } else { - res = Float.parseFloat(val.toString()); - } - } else if ("boolean".equals(esType)) { - if (val instanceof Boolean) { - res = val; - } else if (val instanceof Number) { - int v = ((Number) val).intValue(); - res = v != 0; - } else { - res = Boolean.parseBoolean(val.toString()); - } - } else if ("date".equals(esType)) { - if (val instanceof java.sql.Time) { - DateTime dateTime = new DateTime(((java.sql.Time) val).getTime()); - if (dateTime.getMillisOfSecond() != 0) { - res = dateTime.toString("HH:mm:ss.SSS"); + switch (esType) { + case "integer": + if (val instanceof Number) { + res = ((Number) val).intValue(); } else { - res = dateTime.toString("HH:mm:ss"); + res = Integer.parseInt(val.toString()); } - } else if (val instanceof java.sql.Timestamp) { - DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime()); - if (dateTime.getMillisOfSecond() != 0) { - res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); + break; + case "long": + if (val instanceof Number) { + res = ((Number) val).longValue(); } else { - res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); + res = Long.parseLong(val.toString()); } - } else if (val instanceof java.sql.Date || val instanceof Date) { - DateTime dateTime; - if (val instanceof java.sql.Date) { - dateTime = new DateTime(((java.sql.Date) val).getTime()); + break; + case "short": + if (val instanceof Number) { + res = ((Number) val).shortValue(); } else { - dateTime = new DateTime(((Date) val).getTime()); + res = Short.parseShort(val.toString()); } - if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0 - && dateTime.getMillisOfSecond() == 0) { - res = dateTime.toString("yyyy-MM-dd"); + break; + case "byte": + if (val instanceof Number) { + res = ((Number) val).byteValue(); } else { + res = Byte.parseByte(val.toString()); + } + break; + case "double": + if (val instanceof Number) { + res = ((Number) val).doubleValue(); + } else { + res = Double.parseDouble(val.toString()); + } + break; + case "float": + case "half_float": + case "scaled_float": + if (val instanceof Number) { + res = ((Number) val).floatValue(); + } else { + res = Float.parseFloat(val.toString()); + } + break; + case "boolean": + if (val instanceof Boolean) { + res = val; + } else if (val instanceof Number) { + int v = ((Number) val).intValue(); + res = v != 0; + } else { + res = Boolean.parseBoolean(val.toString()); + } + break; + case "date": + if (val instanceof java.sql.Time) { + DateTime dateTime = new DateTime(((java.sql.Time) val).getTime()); + if (dateTime.getMillisOfSecond() != 0) { + res = dateTime.toString("HH:mm:ss.SSS"); + } else { + res = dateTime.toString("HH:mm:ss"); + } + } else if (val instanceof java.sql.Timestamp) { + DateTime dateTime = new DateTime(((java.sql.Timestamp) val).getTime()); if (dateTime.getMillisOfSecond() != 0) { res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); } else { res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); } - } - } else if (val instanceof Long) { - DateTime dateTime = new DateTime(((Long) val).longValue()); - if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0 - && dateTime.getMillisOfSecond() == 0) { - res = dateTime.toString("yyyy-MM-dd"); - } else if (dateTime.getMillisOfSecond() != 0) { - res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); - } else { - res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); - } - } else if (val instanceof String) { - String v = ((String) val).trim(); - if (v.length() > 18 && v.charAt(4) == '-' && v.charAt(7) == '-' && v.charAt(10) == ' ' - && v.charAt(13) == ':' && v.charAt(16) == ':') { - String dt = v.substring(0, 10) + "T" + v.substring(11); - Date date = Util.parseDate(dt); - if (date != null) { - DateTime dateTime = new DateTime(date); + } else if (val instanceof java.sql.Date || val instanceof Date) { + DateTime dateTime; + if (val instanceof java.sql.Date) { + dateTime = new DateTime(((java.sql.Date) val).getTime()); + } else { + dateTime = new DateTime(((Date) val).getTime()); + } + if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0 + && dateTime.getMillisOfSecond() == 0) { + res = dateTime.toString("yyyy-MM-dd"); + } else { if (dateTime.getMillisOfSecond() != 0) { res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); } else { res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); } } - } else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') { - Date date = Util.parseDate(v); - if (date != null) { - DateTime dateTime = new DateTime(date); + } else if (val instanceof Long) { + DateTime dateTime = new DateTime(((Long) val).longValue()); + if (dateTime.getHourOfDay() == 0 && dateTime.getMinuteOfHour() == 0 && dateTime.getSecondOfMinute() == 0 + && dateTime.getMillisOfSecond() == 0) { res = dateTime.toString("yyyy-MM-dd"); + } else if (dateTime.getMillisOfSecond() != 0) { + res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); + } else { + res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); + } + } else if (val instanceof String) { + String v = ((String) val).trim(); + if (v.length() > 18 && v.charAt(4) == '-' && v.charAt(7) == '-' && v.charAt(10) == ' ' + && v.charAt(13) == ':' && v.charAt(16) == ':') { + String dt = v.substring(0, 10) + "T" + v.substring(11); + Date date = Util.parseDate(dt); + if (date != null) { + DateTime dateTime = new DateTime(date); + if (dateTime.getMillisOfSecond() != 0) { + res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss.SSS" + Util.timeZone); + } else { + res = dateTime.toString("yyyy-MM-dd'T'HH:mm:ss" + Util.timeZone); + } + } + } else if (v.length() == 10 && v.charAt(4) == '-' && v.charAt(7) == '-') { + Date date = Util.parseDate(v); + if (date != null) { + DateTime dateTime = new DateTime(date); + res = dateTime.toString("yyyy-MM-dd"); + } } } - } - } else if ("binary".equals(esType)) { - if (val instanceof byte[]) { - Base64 base64 = new Base64(); - res = base64.encodeAsString((byte[]) val); - } else if (val instanceof Blob) { - byte[] b = blobToBytes((Blob) val); - Base64 base64 = new Base64(); - res = base64.encodeAsString(b); - } else if (val instanceof String) { - // 对应canal中的单字节编码 - byte[] b = ((String) val).getBytes(StandardCharsets.ISO_8859_1); - Base64 base64 = new Base64(); - res = base64.encodeAsString(b); - } - } else if ("geo_point".equals(esType)) { - if (!(val instanceof String)) { - logger.error("es type is geo_point, but source type is not String"); - return val; - } - - if (!((String) val).contains(",")) { - logger.error("es type is geo_point, source value not contains ',' separator"); - return val; - } - - String[] point = ((String) val).split(","); - Map location = new HashMap<>(); - location.put("lat", Double.valueOf(point[0].trim())); - location.put("lon", Double.valueOf(point[1].trim())); - return location; - } else if ("array".equals(esType)) { - if ("".equals(val.toString().trim())) { - res = new ArrayList<>(); - } else { - String value = val.toString(); - String separator = ","; - if (!value.contains(",")) { - if (value.contains(";")) { - separator = ";"; - } else if (value.contains("|")) { - separator = "\\|"; - } else if (value.contains("-")) { - separator = "-"; - } + break; + case "binary": + if (val instanceof byte[]) { + Base64 base64 = new Base64(); + res = base64.encodeAsString((byte[]) val); + } else if (val instanceof Blob) { + byte[] b = blobToBytes((Blob) val); + Base64 base64 = new Base64(); + res = base64.encodeAsString(b); + } else if (val instanceof String) { + // 对应canal中的单字节编码 + byte[] b = ((String) val).getBytes(StandardCharsets.ISO_8859_1); + Base64 base64 = new Base64(); + res = base64.encodeAsString(b); } - String[] values = value.split(separator); - return Arrays.asList(values); - } - } else if ("object".equals(esType)) { - if ("".equals(val.toString().trim())) { - res = new HashMap<>(); - } else { - res = JSON.parseObject(val.toString(), Map.class); - } - } else { - // 其他类全以字符串处理 - res = val.toString(); + break; + case "geo_point": + if (!(val instanceof String)) { + logger.error("es type is geo_point, but source type is not String"); + return val; + } + + if (!((String) val).contains(",")) { + logger.error("es type is geo_point, source value not contains ',' separator"); + return val; + } + + String[] point = ((String) val).split(","); + Map location = new HashMap<>(); + location.put("lat", Double.valueOf(point[0].trim())); + location.put("lon", Double.valueOf(point[1].trim())); + return location; + case "array": + if ("".equals(val.toString().trim())) { + res = new ArrayList<>(); + } else { + String value = val.toString(); + String separator = ","; + if (!value.contains(",")) { + if (value.contains(";")) { + separator = ";"; + } else if (value.contains("|")) { + separator = "\\|"; + } else if (value.contains("-")) { + separator = "-"; + } + } + String[] values = value.split(separator); + return Arrays.asList(values); + } + break; + case "object": + if ("".equals(val.toString().trim())) { + res = new HashMap<>(); + } else { + res = JSON.parseObject(val.toString(), Map.class); + } + break; + default: + // 其他类全以字符串处理 + res = val.toString(); + break; } return res; diff --git a/client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduEtlService.java b/client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduEtlService.java index 5560db41..e4f3afad 100644 --- a/client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduEtlService.java +++ b/client-adapter/kudu/src/main/java/com/alibaba/otter/canal/client/adapter/kudu/service/KuduEtlService.java @@ -88,45 +88,40 @@ public class KuduEtlService extends AbstractEtlService { Map data = new HashMap<>(); for (Map.Entry entry : columnsMap.entrySet()) { String mysqlColumnName = entry.getKey();// mysql字段名 - String kuduColumnName = entry.getValue();// kudu字段名 - if (kuduColumnName == null) { - kuduColumnName = mysqlColumnName; - } - Object value = rs.getObject(kuduColumnName); - if (value != null) { - data.put(kuduColumnName, value); - } else { - data.put(kuduColumnName, null); - } - } - dataList.add(data); - idx++; - impCount.incrementAndGet(); - if (logger.isDebugEnabled()) { - logger.debug("successful import count:" + impCount.get()); - } - if (idx % kuduMapping.getCommitBatch() == 0) { - kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList); - dataList.clear(); - } - } - if (!dataList.isEmpty()) { - kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList); - } - return true; + String kuduColumnName = entry.getValue();// kudu字段名 + if (kuduColumnName == null) { + kuduColumnName = mysqlColumnName; + } + Object value = rs.getObject(kuduColumnName); + if (value != null) { + data.put(kuduColumnName, value); + } else { + data.put(kuduColumnName, null); + } + } + dataList.add(data); + idx++; + impCount.incrementAndGet(); + if (logger.isDebugEnabled()) { + logger.debug("successful import count:" + impCount.get()); + } + if (idx % kuduMapping.getCommitBatch() == 0) { + kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList); + dataList.clear(); + } + } + if (!dataList.isEmpty()) { + kuduTemplate.upsert(kuduMapping.getTargetTable(), dataList); + } + return true; - } catch (SQLException e) { - e.printStackTrace(); - logger.error(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage(), e); - errMsg.add(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage()); - return false; - } catch (KuduException e) { - e.printStackTrace(); - logger.error(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage(), e); - errMsg.add(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage()); - return false; - } -} ); + } catch (SQLException | KuduException e) { + e.printStackTrace(); + logger.error(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage(), e); + errMsg.add(kuduMapping.getTargetTable() + " etl failed! ==>" + e.getMessage()); + return false; + } + }); return true; } catch (Exception e) { logger.error(e.getMessage(), e); diff --git a/common/src/main/java/com/alibaba/otter/canal/common/utils/AddressUtils.java b/common/src/main/java/com/alibaba/otter/canal/common/utils/AddressUtils.java index e380cf3b..53bc8e35 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/utils/AddressUtils.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/utils/AddressUtils.java @@ -18,20 +18,11 @@ public class AddressUtils { private static final Pattern IP_PATTERN = Pattern.compile("[0-9]{1,3}(\\.[0-9]{1,3}){3,}"); public static boolean isAvailablePort(int port) { - ServerSocket ss = null; - try { - ss = new ServerSocket(port); + try (ServerSocket ss = new ServerSocket(port)) { ss.bind(null); return true; } catch (IOException e) { return false; - } finally { - if (ss != null) { - try { - ss.close(); - } catch (IOException e) { - } - } } } diff --git a/common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java b/common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java index d4618b9f..456d9688 100644 --- a/common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java +++ b/common/src/main/java/com/alibaba/otter/canal/common/utils/JsonUtils.java @@ -71,14 +71,11 @@ public class JsonUtils { */ public static String marshalToString(Object obj, String... fliterFields) { final List propertyFliters = Arrays.asList(fliterFields); - SerializeWriter out = new SerializeWriter(); - try { + try (SerializeWriter out = new SerializeWriter()) { JSONSerializer serializer = new JSONSerializer(out); serializer.getPropertyFilters().add((source, name, value) -> !propertyFliters.contains(name)); serializer.write(obj); return out.toString(); - } finally { - out.close(); } } diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java index 1a370cf1..d14f5a14 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/filter/AviaterRegexFilter.java @@ -51,7 +51,7 @@ public class AviaterRegexFilter { // 对pattern按照从长到短的排序 // 因为 foo|foot 匹配 foot 会出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot // 的长度不一样 - Collections.sort(list, COMPARATOR); + list.sort(COMPARATOR); // 对pattern进行头尾完全匹配 list = completionPattern(list); this.pattern = StringUtils.join(list, PATTERN_SPLIT); @@ -88,13 +88,7 @@ public class AviaterRegexFilter { @Override public int compare(String str1, String str2) { - if (str1.length() > str2.length()) { - return -1; - } else if (str1.length() < str2.length()) { - return 1; - } else { - return 0; - } + return Integer.compare(str2.length(), str1.length()); } } diff --git a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/util/CanalMessageSerializerUtil.java b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/util/CanalMessageSerializerUtil.java index 4e7929b6..5fac0e67 100644 --- a/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/util/CanalMessageSerializerUtil.java +++ b/connector/core/src/main/java/com/alibaba/otter/canal/connector/core/util/CanalMessageSerializerUtil.java @@ -34,8 +34,8 @@ public class CanalMessageSerializerUtil { messageSize += CodedOutputStream.computeInt64Size(1, data.getId()); int dataSize = 0; - for (int i = 0; i < rowEntries.size(); i++) { - dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry); } messageSize += dataSize; messageSize += 1 * rowEntries.size(); @@ -53,8 +53,8 @@ public class CanalMessageSerializerUtil { output.writeRawVarint32(messageSize); // message output.writeInt64(1, data.getId()); - for (int i = 0; i < rowEntries.size(); i++) { - output.writeBytes(2, rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + output.writeBytes(2, rowEntry); } output.checkNoSpaceLeft(); return body; diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java index d9b6355b..21b8419e 100644 --- a/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java +++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/FetcherPerformanceTest.java @@ -12,8 +12,7 @@ import com.taobao.tddl.dbsync.binlog.LogEvent; public class FetcherPerformanceTest { public static void main(String args[]) { - DirectLogFetcher fetcher = new DirectLogFetcher(); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher()) { Class.forName("com.mysql.jdbc.Driver"); Connection connection = DriverManager.getConnection("jdbc:mysql://127.0.0.1:3306", "root", "hello"); Statement statement = connection.createStatement(); @@ -40,11 +39,6 @@ public class FetcherPerformanceTest { } } catch (Exception e) { e.printStackTrace(); - } finally { - try { - fetcher.close(); - } catch (IOException e) { - } } } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java index 29cee67c..b50809e0 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/admin/CanalAdminController.java @@ -192,7 +192,7 @@ public class CanalAdminController implements CanalAdmin { Collection files = org.apache.commons.io.FileUtils.listFiles(new File("../logs/canal/"), TrueFileFilter.TRUE, TrueFileFilter.TRUE); - List names = files.stream().map(f -> f.getName()).collect(Collectors.toList()); + List names = files.stream().map(File::getName).collect(Collectors.toList()); return Joiner.on(",").join(names); } @@ -206,7 +206,7 @@ public class CanalAdminController implements CanalAdmin { Collection files = org.apache.commons.io.FileUtils.listFiles(new File("../logs/" + destination + "/"), TrueFileFilter.TRUE, TrueFileFilter.TRUE); - List names = files.stream().map(f -> f.getName()).collect(Collectors.toList()); + List names = files.stream().map(File::getName).collect(Collectors.toList()); return Joiner.on(",").join(names); } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java index 23629559..eaa8afbc 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/MysqlGTIDSet.java @@ -89,8 +89,8 @@ public class MysqlGTIDSet implements GTIDSet { // 存在多个GTID时会有回车符 String[] uuidStrs = gtidData.replaceAll("\n", "").split(","); m = new HashMap<>(uuidStrs.length); - for (int i = 0; i < uuidStrs.length; i++) { - UUIDSet uuidSet = UUIDSet.parse(uuidStrs[i]); + for (String uuidStr : uuidStrs) { + UUIDSet uuidSet = UUIDSet.parse(uuidStr); m.put(uuidSet.SID.toString(), uuidSet); } } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java index e6908a92..0eaea9cd 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/packets/UUIDSet.java @@ -80,7 +80,7 @@ public class UUIDSet { if (equals(o)) { return 1; } - return start > o.start ? 1 : (start == o.start ? 0 : -1); + return Long.compare(start, o.start); } } diff --git a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java index de1d3710..276040d7 100644 --- a/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java +++ b/filter/src/main/java/com/alibaba/otter/canal/filter/aviater/AviaterRegexFilter.java @@ -53,7 +53,7 @@ public class AviaterRegexFilter implements CanalEventFilter { // 对pattern按照从长到短的排序 // 因为 foo|foot 匹配 foot 会出错，原因是 foot 匹配了 foo 之后，会返回 foo，但是 foo 的长度和 foot // 的长度不一样 - Collections.sort(list, COMPARATOR); + list.sort(COMPARATOR); // 对pattern进行头尾完全匹配 list = completionPattern(list); this.pattern = StringUtils.join(list, PATTERN_SPLIT); @@ -90,13 +90,7 @@ public class AviaterRegexFilter implements CanalEventFilter { @Override public int compare(String str1, String str2) { - if (str1.length() > str2.length()) { - return -1; - } else if (str1.length() < str2.length()) { - return 1; - } else { - return 0; - } + return Integer.compare(str2.length(), str1.length()); } } diff --git a/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java b/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java index 255b98a7..a9e65b24 100644 --- a/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java +++ b/meta/src/main/java/com/alibaba/otter/canal/meta/ZooKeeperMetaManager.java @@ -191,7 +191,7 @@ public class ZooKeeperMetaManager extends AbstractCanalLifeCycle implements Cana } // 找到最小的Id - ArrayList batchIds = new ArrayList(nodes.size()); + ArrayList batchIds = new ArrayList<>(nodes.size()); for (String batchIdString : nodes) { batchIds.add(Long.valueOf(batchIdString)); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java index 411a19d6..457b5d3b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/LocalBinLogConnection.java @@ -81,10 +81,9 @@ public class LocalBinLogConnection implements ErosaConnection { public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException { File current = new File(directory, binlogfilename); - FileLogFetcher fetcher = new FileLogFetcher(bufferSize); - LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); - LogContext context = new LogContext(); - try { + try (FileLogFetcher fetcher = new FileLogFetcher(bufferSize)) { + LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); + LogContext context = new LogContext(); fetcher.open(current, binlogPosition); context.setLogPosition(new LogPosition(binlogfilename, binlogPosition)); while (running) { @@ -129,10 +128,6 @@ public class LocalBinLogConnection implements ErosaConnection { } } catch (InterruptedException e) { logger.warn("LocalBinLogConnection dump interrupted"); - } finally { - if (fetcher != null) { - fetcher.close(); - } } } @@ -226,10 +221,9 @@ public class LocalBinLogConnection implements ErosaConnection { throw new CanalParseException("binlog:" + binlogfilename + " is not found"); } - FileLogFetcher fetcher = new FileLogFetcher(bufferSize); - LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); - LogContext context = new LogContext(); - try { + try (FileLogFetcher fetcher = new FileLogFetcher(bufferSize)) { + LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); + LogContext context = new LogContext(); fetcher.open(current, binlogPosition); context.setLogPosition(new LogPosition(binlogfilename, binlogPosition)); while (running) { @@ -273,10 +267,6 @@ public class LocalBinLogConnection implements ErosaConnection { } } catch (InterruptedException e) { logger.warn("LocalBinLogConnection dump interrupted"); - } finally { - if (fetcher != null) { - fetcher.close(); - } } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index 59dea37a..7d04b2be 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -191,8 +191,7 @@ public class MysqlConnection implements ErosaConnection { loadBinlogChecksum(); sendBinlogDumpGTID(gtidSet); - DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize())) { fetcher.start(connector.getChannel()); LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT); LogContext context = new LogContext(); @@ -212,8 +211,6 @@ public class MysqlConnection implements ErosaConnection { break; } } - } finally { - fetcher.close(); } } @@ -229,8 +226,7 @@ public class MysqlConnection implements ErosaConnection { sendBinlogDump(binlogfilename, binlogPosition); ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this); ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum); - DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize())) { fetcher.start(connector.getChannel()); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); @@ -240,8 +236,6 @@ public class MysqlConnection implements ErosaConnection { break; } } - } finally { - fetcher.close(); } } @@ -257,8 +251,7 @@ public class MysqlConnection implements ErosaConnection { sendBinlogDumpGTID(gtidSet); ((MysqlMultiStageCoprocessor) coprocessor).setConnection(this); ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum); - DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize()); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize())) { fetcher.start(connector.getChannel()); while (fetcher.fetch()) { accumulateReceivedBytes(fetcher.limit()); @@ -268,8 +261,6 @@ public class MysqlConnection implements ErosaConnection { break; } } - } finally { - fetcher.close(); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 517a8247..49246ffd 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -245,18 +245,6 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE if (haController != null && haController instanceof HeartBeatCallback) { ((HeartBeatCallback) haController).onSuccess(costTime); } - } catch (SocketTimeoutException e) { - if (haController != null && haController instanceof HeartBeatCallback) { - ((HeartBeatCallback) haController).onFailed(e); - } - reconnect = true; - logger.warn("connect failed by ", e); - } catch (IOException e) { - if (haController != null && haController instanceof HeartBeatCallback) { - ((HeartBeatCallback) haController).onFailed(e); - } - reconnect = true; - logger.warn("connect failed by ", e); } catch (Throwable e) { if (haController != null && haController instanceof HeartBeatCallback) { ((HeartBeatCallback) haController).onFailed(e); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java index 222d762a..76cda929 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/DirectLogFetcher.java @@ -150,11 +150,7 @@ public class DirectLogFetcher extends LogFetcher { close(); /* Do cleanup */ logger.error("Socket timeout expired, closing connection", e); throw e; - } catch (InterruptedIOException e) { - close(); /* Do cleanup */ - logger.info("I/O interrupted while reading from client socket", e); - throw e; - } catch (ClosedByInterruptException e) { + } catch (InterruptedIOException | ClosedByInterruptException e) { close(); /* Do cleanup */ logger.info("I/O interrupted while reading from client socket", e); throw e; diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java index ca09dccb..08b1b164 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/local/BinLogFileQueue.java @@ -230,7 +230,7 @@ public class BinLogFileQueue { } }, null)); // 排一下序列 - Collections.sort(files, Comparator.comparing(File::getName)); + files.sort(Comparator.comparing(File::getName)); return files; } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java index e8e715a2..3e2edc55 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/HttpHelper.java @@ -56,8 +56,7 @@ public class HttpHelper { HttpGet httpGet = new HttpGet(uri); HttpClientContext context = HttpClientContext.create(); context.setRequestConfig(config); - CloseableHttpResponse response = httpclient.execute(httpGet, context); - try { + try (CloseableHttpResponse response = httpclient.execute(httpGet, context)) { int statusCode = response.getStatusLine().getStatusCode(); long end = System.currentTimeMillis(); long cost = end - start; @@ -69,10 +68,9 @@ public class HttpHelper { } else { String errorMsg = EntityUtils.toString(response.getEntity()); throw new RuntimeException("requestGet remote error, url=" + uri.toString() + ", code=" + statusCode - + ", error msg=" + errorMsg); + + ", error msg=" + errorMsg); } } finally { - response.close(); httpGet.releaseConnection(); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java index f7c75c37..47d919a6 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/index/PeriodMixedLogPositionManager.java @@ -83,7 +83,7 @@ public class PeriodMixedLogPositionManager extends AbstractLogPositionManager { // 启动定时工作任务 executorService.scheduleAtFixedRate(() -> { - List tasks = new ArrayList(persistTasks); + List tasks = new ArrayList<>(persistTasks); for (String destination : tasks) { try { // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次 diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogEventPerformanceTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogEventPerformanceTest.java index 6068fcf2..fc27ac60 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogEventPerformanceTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogEventPerformanceTest.java @@ -22,8 +22,7 @@ public class MysqlBinlogEventPerformanceTest { protected static Charset charset = Charset.forName("utf-8"); public static void main(String args[]) { - DirectLogFetcher fetcher = new DirectLogFetcher(); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher()) { MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello"); connector.connect(); updateSettings(connector); @@ -49,11 +48,6 @@ public class MysqlBinlogEventPerformanceTest { } } catch (Exception e) { e.printStackTrace(); - } finally { - try { - fetcher.close(); - } catch (IOException e) { - } } } diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java index 99431395..58a9a9a4 100644 --- a/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java +++ b/parse/src/test/java/com/alibaba/otter/canal/parse/MysqlBinlogParsePerformanceTest.java @@ -34,8 +34,7 @@ public class MysqlBinlogParsePerformanceTest { protected static Charset charset = Charset.forName("utf-8"); public static void main(String args[]) { - DirectLogFetcher fetcher = new DirectLogFetcher(); - try { + try (DirectLogFetcher fetcher = new DirectLogFetcher()) { MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello"); connector.connect(); updateSettings(connector); @@ -45,9 +44,7 @@ public class MysqlBinlogParsePerformanceTest { Thread thread = new Thread(() -> { try { consumer(buffer); - } catch (IOException e) { - e.printStackTrace(); - } catch (InterruptedException e) { + } catch (IOException | InterruptedException e) { e.printStackTrace(); } }); @@ -59,11 +56,6 @@ public class MysqlBinlogParsePerformanceTest { } } catch (Exception e) { e.printStackTrace(); - } finally { - try { - fetcher.close(); - } catch (IOException e) { - } } } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java index 3fa691ad..cba133e4 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/Message.java @@ -18,24 +18,24 @@ public class Message implements Serializable { private static final long serialVersionUID = 1234034768477580009L; private long id; - private List entries = new ArrayList(); + private List entries = new ArrayList<>(); // row data for performance, see: // https://github.com/alibaba/canal/issues/726 private boolean raw = true; - private List rawEntries = new ArrayList(); + private List rawEntries = new ArrayList<>(); public Message(long id, List entries){ this.id = id; - this.entries = entries == null ? new ArrayList() : entries; + this.entries = entries == null ? new ArrayList<>() : entries; this.raw = false; } public Message(long id, boolean raw, List entries){ this.id = id; if (raw) { - this.rawEntries = entries == null ? new ArrayList() : entries; + this.rawEntries = entries == null ? new ArrayList<>() : entries; } else { - this.entries = entries == null ? new ArrayList() : entries; + this.entries = entries == null ? new ArrayList<>() : entries; } this.raw = raw; } diff --git a/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java b/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java index 43c45c11..268cc6db 100644 --- a/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java +++ b/protocol/src/main/java/com/alibaba/otter/canal/protocol/SecurityUtil.java @@ -84,8 +84,8 @@ public class SecurityUtil { */ public static String byte2HexStr(byte[] b) { StringBuilder hs = new StringBuilder(); - for (int n = 0; n < b.length; n++) { - String hex = (Integer.toHexString(b[n] & 0XFF)); + for (byte value : b) { + String hex = (Integer.toHexString(value & 0XFF)); if (hex.length() == 1) { hs.append("0" + hex); } else { diff --git a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java index aa9eef7a..f553ddfb 100644 --- a/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java +++ b/server/src/main/java/com/alibaba/otter/canal/server/netty/handler/SessionHandler.java @@ -161,8 +161,8 @@ public class SessionHandler extends SimpleChannelHandler { messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId()); int dataSize = 0; - for (int i = 0; i < rowEntries.size(); i++) { - dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry); } messageSize += dataSize; messageSize += 1 * rowEntries.size(); @@ -194,8 +194,8 @@ public class SessionHandler extends SimpleChannelHandler { output.writeRawVarint32(messageSize); // message output.writeInt64(1, message.getId()); - for (int i = 0; i < rowEntries.size(); i++) { - output.writeBytes(2, rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + output.writeBytes(2, rowEntry); } output.checkNoSpaceLeft(); NettyUtils.write(ctx.getChannel(), body, new ChannelFutureAggregator(get.getDestination(), diff --git a/server/src/test/java/com/alibaba/otter/canal/server/ProtocolTest.java b/server/src/test/java/com/alibaba/otter/canal/server/ProtocolTest.java index 6e122266..da479e20 100644 --- a/server/src/test/java/com/alibaba/otter/canal/server/ProtocolTest.java +++ b/server/src/test/java/com/alibaba/otter/canal/server/ProtocolTest.java @@ -63,8 +63,8 @@ public class ProtocolTest { messageSize += com.google.protobuf.CodedOutputStream.computeInt64Size(1, message.getId()); int dataSize = 0; - for (int i = 0; i < rowEntries.size(); i++) { - dataSize += com.google.protobuf.CodedOutputStream.computeBytesSizeNoTag(rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + dataSize += CodedOutputStream.computeBytesSizeNoTag(rowEntry); } messageSize += dataSize; messageSize += 1 * rowEntries.size(); @@ -82,8 +82,8 @@ public class ProtocolTest { output.writeRawVarint32(messageSize); // message output.writeInt64(1, message.getId()); - for (int i = 0; i < rowEntries.size(); i++) { - output.writeBytes(2, rowEntries.get(i)); + for (ByteString rowEntry : rowEntries) { + output.writeBytes(2, rowEntry); } output.checkNoSpaceLeft(); diff --git a/store/src/main/java/com/alibaba/otter/canal/store/model/Events.java b/store/src/main/java/com/alibaba/otter/canal/store/model/Events.java index aa3b6af3..b9bf778c 100644 --- a/store/src/main/java/com/alibaba/otter/canal/store/model/Events.java +++ b/store/src/main/java/com/alibaba/otter/canal/store/model/Events.java @@ -20,7 +20,7 @@ public class Events implements Serializable { private static final long serialVersionUID = -7337454954300706044L; private PositionRange positionRange = new PositionRange(); - private List events = new ArrayList(); + private List events = new ArrayList<>(); public List getEvents() { return events; diff --git a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMemBatchTest.java b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMemBatchTest.java index 4fcab3c3..87fa015e 100644 --- a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMemBatchTest.java +++ b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMemBatchTest.java @@ -81,9 +81,7 @@ public class MemoryEventStoreMemBatchTest extends MemoryEventStoreBase { try { result = eventStore.put(buildEvent("1", 1L, 1L + bufferSize), 1000L, TimeUnit.MILLISECONDS); - } catch (CanalStoreException e) { - Assert.fail(e.getMessage()); - } catch (InterruptedException e) { + } catch (CanalStoreException | InterruptedException e) { Assert.fail(e.getMessage()); } @@ -162,8 +160,7 @@ public class MemoryEventStoreMemBatchTest extends MemoryEventStoreBase { Assert.assertTrue(entrys.getEvents().size() == batchSize); Assert.assertEquals(position, entrys.getPositionRange().getStart()); Assert.assertEquals(position, entrys.getPositionRange().getEnd()); - } catch (CanalStoreException e) { - } catch (InterruptedException e) { + } catch (CanalStoreException | InterruptedException e) { } ExecutorService executor = Executors.newFixedThreadPool(1); diff --git a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMultiThreadTest.java b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMultiThreadTest.java index 1a06e8f1..76dcedea 100644 --- a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMultiThreadTest.java +++ b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStoreMultiThreadTest.java @@ -109,8 +109,7 @@ public class MemoryEventStoreMultiThreadTest extends MemoryEventStoreBase { } try { eventStore.put(event); - } catch (CanalStoreException e) { - } catch (InterruptedException e) { + } catch (CanalStoreException | InterruptedException e) { } } } diff --git a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStorePutAndGetTest.java b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStorePutAndGetTest.java index 473d4582..df80c073 100644 --- a/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStorePutAndGetTest.java +++ b/store/src/test/java/com/alibaba/otter/canal/store/memory/buffer/MemoryEventStorePutAndGetTest.java @@ -64,9 +64,7 @@ public class MemoryEventStorePutAndGetTest extends MemoryEventStoreBase { try { result = eventStore.put(buildEvent("1", 1L, 1L + bufferSize), 1000L, TimeUnit.MILLISECONDS); - } catch (CanalStoreException e) { - Assert.fail(e.getMessage()); - } catch (InterruptedException e) { + } catch (CanalStoreException | InterruptedException e) { Assert.fail(e.getMessage()); } @@ -140,8 +138,7 @@ public class MemoryEventStorePutAndGetTest extends MemoryEventStoreBase { Assert.assertTrue(entrys.getEvents().size() == batchSize); Assert.assertEquals(position, entrys.getPositionRange().getStart()); Assert.assertEquals(position, entrys.getPositionRange().getEnd()); - } catch (CanalStoreException e) { - } catch (InterruptedException e) { + } catch (CanalStoreException | InterruptedException e) { } ExecutorService executor = Executors.newFixedThreadPool(1);