chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复测试类报错 (#4516)

Browse Source 
* optimize YAML config loader (#4332)

* fix bug BASE TABLE as table name (#4217)

* fix issues#4328 (#4329)

* docs: add nodejs canal client support (#4260)

Co-authored-by: zhangxunwei <zhangxunwei@dxy.cn>

* fix destination not encoded (#4279)

* 修复Canal指定时间戳启动失效，总是从最新的点位开始同步问题 (#4348)

* 支持用户自定义的CanalAlarmHandler

* RowsLogEvent增加对TableMapLogEvent判空检查，防止NPE异常

* 修复Canal指定时间戳启动失效，总是从最新的点位开始同步问题, issue: #4347

Co-authored-by: 云时 <mingya.wmy@alibaba-inc.com>

* fixed 4334 , support jdk8/jdk11

* fixed issue #4266 , typo

* fixed issue #4243 , support auto register for cluster = null

* fixed issue #4225 , support mysql version >= 8.0.26 heartbeat v2

* fixed issue #4308 , support query_log_event for maraiadb 10.10.1

* ignore compression event

* support jdk11

* support druid 1.2.12

* fixed mariadb 10.x

* sync canal-template.properties

* update fastjson & druid version (#4406)

* 修复升级2.0.4导致兼容的问题

* update druid & fastjson version

* update fastjson version

* update druid & fastjson version

* meta.dat文件数据丢失 (#4397)

* update fastjson & druid version (#4438)

* 局部变量线程安全，优先使用StringBuilder替换StringBuffer (#4472)

Co-authored-by: 夏亮 <xialiang@newerabc.com>

* 1. CanalController stop 需要同时将 embededCanalServer.stop (#4477)

2. ServerRunningMonitor 线程池未正常回收，线程池管理与 start/stop保持一致

* vuln-fix: Use HTTPS instead of HTTP to resolve dependencies (#4437)

This fixes a security vulnerability in this project where the `build.gradle`
files were configuring Gradle to resolve dependencies over HTTP instead of
HTTPS.

Weakness: CWE-829: Inclusion of Functionality from Untrusted Control Sphere
Severity: High
CVSSS: 8.1
Detection: OpenRewrite

Reported-by: Jonathan Leitschuh <Jonathan.Leitschuh@gmail.com>
Signed-off-by: Jonathan Leitschuh <Jonathan.Leitschuh@gmail.com>

Bug-tracker: https://github.com/JLLeitschuh/security-research/issues/9


Co-authored-by: Moderne <team@moderne.io>

Co-authored-by: Moderne <team@moderne.io>

* fix_ETL同步mysql关键字报错 (#4346)

Co-authored-by: foleyang <foleyang@cogobuy.com>

* optimize code

* use compact BigDecimal

* performance optimize , 1. cache string names 2. cache Charset

* performance optimize ,cache integer/long valueof

* 测试类报错

Co-authored-by: gongchangyou <gongchangyou@gmail.com>
Co-authored-by: ChanaLii <316529035@qq.com>
Co-authored-by: zhangxunwei <zhangxunweia@gmail.com>
Co-authored-by: zhangxunwei <zhangxunwei@dxy.cn>
Co-authored-by: tianpeidong <39491687+tianpeidong@users.noreply.github.com>
Co-authored-by: dataccs <dataccs@163.com>
Co-authored-by: 云时 <mingya.wmy@alibaba-inc.com>
Co-authored-by: jianghang.loujh <jianghang.loujh@alibaba-inc.com>
Co-authored-by: 温绍锦 <shaojin.wensj@alibaba-inc.com>
Co-authored-by: noaso <noasoso@gmail.com>
Co-authored-by: HumanPassenger <40585855+HumanPassenger@users.noreply.github.com>
Co-authored-by: 夏亮 <xialiang@newerabc.com>
Co-authored-by: 华仔 <591327356@qq.com>
Co-authored-by: Jonathan Leitschuh <jonathan.leitschuh@gmail.com>
Co-authored-by: Moderne <team@moderne.io>
Co-authored-by: 杰锅不是锅 <1105568074@qq.com>
Co-authored-by: foleyang <foleyang@cogobuy.com>
This commit is contained in:
rewerma
2022-12-06 16:17:01 +08:00
committed by GitHub
parent a96db39420
commit d899981354
89 changed files with 1108 additions and 4473 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/CharsetConversion.java
View File

@@ -21,6 +21,7 @@ public final class CharsetConversion {
protected final String mysqlCharset;
protected final String mysqlCollation;
protected final String javaCharset;
protected final Charset charset;
Entry(final int id, String mysqlCharset, // NL
String mysqlCollation, String javaCharset){
@@ -28,6 +29,7 @@ public final class CharsetConversion {
this.mysqlCharset = mysqlCharset;
this.mysqlCollation = mysqlCollation;
this.javaCharset = javaCharset;
this.charset = Charset.isSupported(javaCharset) ? Charset.forName(javaCharset) : null;
}
}
@@ -390,6 +392,23 @@ public final class CharsetConversion {
}
}
public static Charset getNioCharset(final int id) {
Entry entry = getEntry(id);
if (entry != null) {
if (entry.charset != null) {
return entry.charset;
} else {
logger.warn("Unknown java charset for: id = " + id + ", name = " + entry.mysqlCharset + ", coll = "
+ entry.mysqlCollation);
return null;
}
} else {
logger.warn("Unexpect mysql charset: " + id);
return null;
}
}
public static void main(String[] args) {
for (int i = 0; i < entries.length; i++) {
Entry entry = entries[i];

42
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java
View File

@@ -1,5 +1,7 @@
package com.taobao.tddl.dbsync.binlog;
import java.nio.charset.Charset;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber2;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.appendNumber4;
import static com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer.usecondsToStr;
@@ -57,23 +59,27 @@ public class JsonConversion {
public static final int VALUE_ENTRY_SIZE_LARGE = (1 + LARGE_OFFSET_SIZE);
public static Json_Value parse_value(int type, LogBuffer buffer, long len, String charsetName) {
return parse_value(type, buffer, len, Charset.forName(charsetName));
}
public static Json_Value parse_value(int type, LogBuffer buffer, long len, Charset charset) {
buffer = buffer.duplicate(buffer.position(), (int) len);
switch (type) {
case JSONB_TYPE_SMALL_OBJECT:
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charsetName);
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, false, charset);
case JSONB_TYPE_LARGE_OBJECT:
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charsetName);
return parse_array_or_object(Json_enum_type.OBJECT, buffer, len, true, charset);
case JSONB_TYPE_SMALL_ARRAY:
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charsetName);
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, false, charset);
case JSONB_TYPE_LARGE_ARRAY:
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charsetName);
return parse_array_or_object(Json_enum_type.ARRAY, buffer, len, true, charset);
default:
return parse_scalar(type, buffer, len, charsetName);
return parse_scalar(type, buffer, len, charset);
}
}
private static Json_Value parse_array_or_object(Json_enum_type type, LogBuffer buffer, long len, boolean large,
String charsetName) {
Charset charset) {
long offset_size = large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
if (len < 2 * offset_size) {
throw new IllegalArgumentException("illegal json data");
@@ -100,7 +106,7 @@ public class JsonConversion {
return large ? buffer.getUint32() : buffer.getUint16();
}
private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, String charsetName) {
private static Json_Value parse_scalar(int type, LogBuffer buffer, long len, Charset charset) {
switch (type) {
case JSONB_TYPE_LITERAL:
/* purecov: inspected */
@@ -156,7 +162,7 @@ public class JsonConversion {
}
return new Json_Value(Json_enum_type.STRING, buffer.rewind()
.forward((int) n)
.getFixString((int) str_len, charsetName));
.getFixString((int) str_len, charset));
case JSONB_TYPE_OPAQUE:
/*
* There should always be at least one byte, which tells the
@@ -242,7 +248,7 @@ public class JsonConversion {
this.m_large = large;
}
public String key(int i, String charsetName) {
public String key(int i, Charset charset) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -255,10 +261,10 @@ public class JsonConversion {
// entry, always two
// bytes.
long key_length = m_data.getUint16();
return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charsetName);
return m_data.rewind().forward((int) key_offset).getFixString((int) key_length, charset);
}
public Json_Value element(int i, String charsetName) {
public Json_Value element(int i, Charset charset) {
m_data.rewind();
int offset_size = m_large ? LARGE_OFFSET_SIZE : SMALL_OFFSET_SIZE;
int key_entry_size = m_large ? KEY_ENTRY_SIZE_LARGE : KEY_ENTRY_SIZE_SMALL;
@@ -271,13 +277,13 @@ public class JsonConversion {
int type = m_data.forward(entry_offset).getUint8();
if (type == JSONB_TYPE_INT16 || type == JSONB_TYPE_UINT16 || type == JSONB_TYPE_LITERAL
|| (m_large && (type == JSONB_TYPE_INT32 || type == JSONB_TYPE_UINT32))) {
return parse_scalar(type, m_data, value_entry_size - 1, charsetName);
return parse_scalar(type, m_data, value_entry_size - 1, charset);
}
int value_offset = (int) read_offset_or_size(m_data, m_large);
return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charsetName);
return parse_value(type, m_data.rewind().forward(value_offset), (int) m_length - value_offset, charset);
}
public StringBuilder toJsonString(StringBuilder buf, String charsetName) {
public StringBuilder toJsonString(StringBuilder buf, Charset charset) {
switch (m_type) {
case OBJECT:
buf.append("{");
@@ -285,9 +291,9 @@ public class JsonConversion {
if (i > 0) {
buf.append(", ");
}
buf.append('"').append(key(i, charsetName)).append('"');
buf.append('"').append(key(i, charset)).append('"');
buf.append(": ");
element(i, charsetName).toJsonString(buf, charsetName);
element(i, charset).toJsonString(buf, charset);
}
buf.append("}");
break;
@@ -297,7 +303,7 @@ public class JsonConversion {
if (i > 0) {
buf.append(", ");
}
element(i, charsetName).toJsonString(buf, charsetName);
element(i, charset).toJsonString(buf, charset);
}
buf.append("]");
break;
@@ -398,7 +404,7 @@ public class JsonConversion {
}
buf.append('"').append(text).append('"');
} else {
text = m_data.getFixString((int) m_length, charsetName);
text = m_data.getFixString((int) m_length, charset);
buf.append('"').append(escapse(text)).append('"');
}

10
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonDiffConversion.java
View File

@@ -1,5 +1,6 @@
package com.taobao.tddl.dbsync.binlog;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;
@@ -36,6 +37,11 @@ public class JsonDiffConversion {
public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
String charsetName) {
return print_json_diff(buffer, len, columnName, columnIndex, Charset.forName(charsetName));
}
public static StringBuilder print_json_diff(LogBuffer buffer, long len, String columnName, int columnIndex,
Charset charset) {
int position = buffer.position();
List<String> operation_names = new ArrayList<>();
while (buffer.hasRemaining()) {
@@ -111,14 +117,14 @@ public class JsonDiffConversion {
Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
buffer,
value_length - 1,
charsetName);
charset);
buffer.forward((int) value_length - 1);
// Read value
if (jsonValue.m_type == Json_enum_type.ERROR) {
throw new IllegalArgumentException("parsing json value");
}
StringBuilder jsonBuilder = new StringBuilder();
jsonValue.toJsonString(jsonBuilder, charsetName);
jsonValue.toJsonString(jsonBuilder, charset);
builder.append(jsonBuilder);
}

369
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogBuffer.java
View File

@@ -5,6 +5,8 @@ import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.BitSet;
@@ -16,6 +18,8 @@ import java.util.BitSet;
*/
public class LogBuffer {
static final BigDecimal DECIMAL_ZERO_1_SCALE = BigDecimal.valueOf(0, 1);
static final BigDecimal DECIMAL_ONE_1_SCALE = BigDecimal.valueOf(10, 1);
protected byte[] buffer;
protected int origin, limit;
@@ -1038,29 +1042,27 @@ public class LogBuffer {
}
}
/* default ANSI charset */
public static final String ISO_8859_1 = "ISO-8859-1";
/**
* Return fix length string from buffer.
*/
public final String getFixString(final int pos, final int len) {
return getFixString(pos, len, ISO_8859_1);
return getFixString(pos, len, StandardCharsets.ISO_8859_1);
}
/**
* Return next fix length string from buffer.
*/
public final String getFixString(final int len) {
return getFixString(len, ISO_8859_1);
return getFixString(len, StandardCharsets.ISO_8859_1);
}
/**
* Return fix length string from buffer.
*/
public final String getFixString(final int pos, final int len, String charsetName) {
if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
+ (pos < 0 ? pos : (pos + len)));
public final String getFixString(final int pos, final int len, Charset charset) {
if (pos + len > limit || pos < 0) {
throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
}
final int from = origin + pos;
final int end = from + len;
@@ -1069,19 +1071,16 @@ public class LogBuffer {
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
try {
return new String(buf, from, found - from, charsetName);
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
}
return new String(buf, from, found - from, charset);
}
/**
* Return next fix length string from buffer.
*/
public final String getFixString(final int len, String charsetName) {
if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
+ (position + len - origin));
public final String getFixString(final int len, Charset charset) {
if (position + len > origin + limit) {
throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
final int from = position;
final int end = from + len;
@@ -1090,97 +1089,141 @@ public class LogBuffer {
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
try {
String string = new String(buf, from, found - from, charsetName);
position += len;
return string;
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
String string = new String(buf, from, found - from, charset);
position += len;
return string;
}
public final String getFixName(final int len, Charset charset) {
if (position + len > origin + limit) {
throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
final int from = position;
final int end = from + len;
byte[] buf = buffer;
int found = from;
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
int length = found - from;
String string = null;
if (length <= 16) {
string = NameCache.name(buf, from, length, charset);
}
if (string == null) {
string = new String(buf, from, length, charset);
}
position += len;
return string;
}
public final String getFixName(final int len) {
if (position + len > origin + limit) {
throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
final int from = position;
final int end = from + len;
byte[] buf = buffer;
int found = from;
for (; (found < end) && buf[found] != '\0'; found++)
/* empty loop */;
int length = found - from;
String string = null;
if (length <= 16) {
string = NameCache.name(buf, from, length, StandardCharsets.ISO_8859_1);
}
if (string == null) {
string = new String(buf, from, length, StandardCharsets.ISO_8859_1);
}
position += len;
return string;
}
/**
* Return fix-length string from buffer without null-terminate checking. Fix
* bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
* Return fix-length string from buffer without null-terminate checking.
* Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
*/
public final String getFullString(final int pos, final int len, String charsetName) {
if (pos + len > limit || pos < 0) throw new IllegalArgumentException("limit excceed: "
+ (pos < 0 ? pos : (pos + len)));
try {
return new String(buffer, origin + pos, len, charsetName);
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
public final String getFullString(final int pos, final int len, Charset charset) {
if (pos + len > limit || pos < 0) {
throw new IllegalArgumentException("limit excceed: " + (pos < 0 ? pos : (pos + len)));
}
return new String(buffer, origin + pos, len, charset);
}
/**
* Return next fix-length string from buffer without null-terminate
* checking. Fix bug #17 {@link https
* ://github.com/AlibabaTech/canal/issues/17 }
* checking.
* Fix bug #17 {@link https://github.com/AlibabaTech/canal/issues/17 }
*/
public final String getFullString(final int len, String charsetName) {
if (position + len > origin + limit) throw new IllegalArgumentException("limit excceed: "
+ (position + len - origin));
try {
String string = new String(buffer, position, len, charsetName);
position += len;
return string;
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
public final String getFullString(final int len, Charset charset) {
if (position + len > origin + limit) {
throw new IllegalArgumentException("limit excceed: " + (position + len - origin));
}
String string = new String(buffer, position, len, charset);
position += len;
return string;
}
/**
* Return dynamic length string from buffer.
*/
public final String getString(final int pos) {
return getString(pos, ISO_8859_1);
if (pos >= limit || pos < 0) {
throw new IllegalArgumentException("limit excceed: " + pos);
}
byte[] buf = buffer;
final int len = (0xff & buf[origin + pos]);
if (pos + len + 1 > limit) {
throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
}
return new String(buf, origin + pos + 1, len, StandardCharsets.ISO_8859_1);
}
/**
* Return next dynamic length string from buffer.
*/
public final String getString() {
return getString(ISO_8859_1);
}
/**
* Return dynamic length string from buffer.
*/
public final String getString(final int pos, String charsetName) {
if (pos >= limit || pos < 0) throw new IllegalArgumentException("limit excceed: " + pos);
byte[] buf = buffer;
final int len = (0xff & buf[origin + pos]);
if (pos + len + 1 > limit) throw new IllegalArgumentException("limit excceed: " + (pos + len + 1));
try {
return new String(buf, origin + pos + 1, len, charsetName);
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
if (position >= origin + limit) {
throw new IllegalArgumentException("limit excceed: " + position);
}
}
/**
* Return next dynamic length string from buffer.
*/
public final String getString(String charsetName) {
if (position >= origin + limit) throw new IllegalArgumentException("limit excceed: " + position);
byte[] buf = buffer;
final int len = (0xff & buf[position]);
if (position + len + 1 > origin + limit) throw new IllegalArgumentException("limit excceed: "
+ (position + len + 1 - origin));
try {
String string = new String(buf, position + 1, len, charsetName);
position += len + 1;
return string;
} catch (UnsupportedEncodingException e) {
throw new IllegalArgumentException("Unsupported encoding: " + charsetName, e);
if (position + len + 1 > origin + limit) {
throw new IllegalArgumentException("limit excceed: "
+ (position + len + 1 - origin));
}
String string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
position += len + 1;
return string;
}
public final String getName() {
if (position >= origin + limit) {
throw new IllegalArgumentException("limit excceed: " + position);
}
byte[] buf = buffer;
final int len = (0xff & buf[position]);
if (position + len + 1 > origin + limit) {
throw new IllegalArgumentException("limit excceed: " + (position + len + 1 - origin));
}
String string = NameCache.name(buf, position + 1, len, StandardCharsets.ISO_8859_1);
if (string == null) {
string = new String(buf, position + 1, len, StandardCharsets.ISO_8859_1);
}
position += len + 1;
return string;
}
/**
@@ -1263,6 +1306,15 @@ public class LogBuffer {
throw new IllegalArgumentException("limit excceed: " + (position + binSize - origin));
}
if (precision <= 20 && intg <= 18 && frac <= 9) {
// use compact BigDecimal
BigDecimal compactDecimal = getBigDecimalCompact(precision, scale, intg, frac, intg0x);
if (compactDecimal != null) {
position += binSize;
return compactDecimal;
}
}
BigDecimal decimal = getDecimal0(position, intg, frac, // NL
intg0,
frac0,
@@ -1272,6 +1324,162 @@ public class LogBuffer {
return decimal;
}
/**
* use compact BigDecimal
*/
private BigDecimal getBigDecimalCompact(int precision, int scale, int intg, int frac, int intg0x) {
if (precision > 20 || intg > 18 && frac > 9) {
return null;
}
BigDecimal compactDecimal = null;
long longValue = 0;
int small = 0;
int pos = position;
byte b0 = buffer[position];
final int mask = ((b0 & 0x80) == 0x80) ? 0 : -1;
b0 ^= 0x80;
switch (intg0x) {
case 0:
if (intg == 0) {
longValue = 0;
} else {
longValue = (b0 << 24)
| ((0xff & buffer[pos + 1]) << 16)
| ((0xff & buffer[pos + 2]) << 8)
| (0xff & buffer[pos + 3]);
longValue ^= mask;
pos += 4;
}
break;
case 1:
case 2:
longValue = b0;
longValue ^= mask;
pos += 1;
break;
case 3:
case 4:
longValue = (b0 << 8) | (0xff & buffer[pos + 1]);
longValue ^= mask;
pos += 2;
break;
case 5:
case 6:
longValue = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
longValue ^= mask;
pos += 3;
break;
case 7:
case 8:
longValue = (b0 << 24)
| ((0xff & buffer[pos + 1]) << 16)
| ((0xff & buffer[pos + 2]) << 8)
| (0xff & buffer[pos + 3]);
longValue ^= mask;
pos += 4;
break;
default:
break;
}
if (intg > 9) {
int value = getInt32BE(buffer, pos);
value ^= mask;
longValue *= 1_000_000_000;
longValue += value;
pos += 4;
}
int power = 1;
if (pos != position) {
b0 = buffer[pos];
}
switch (frac) {
case 0:
small = 0;
break;
case 1:
case 2:
power = frac == 1 ? 10 : 100;
small = b0;
small ^= mask;
break;
case 3:
case 4:
power = frac == 3 ? 1_000 : 10_000;
small = (b0 << 8) | (0xff & buffer[pos + 1]);
small ^= mask;
break;
case 5:
case 6:
power = frac == 5 ? 100_000 : 1_000_000;
small = (b0 << 16) | ((0xff & buffer[pos + 1]) << 8) | (0xff & buffer[pos + 2]);
small ^= mask;
break;
case 7:
case 8:
case 9:
power = frac == 7 ? 10_000_000 : frac == 8 ? 100_000_000 : 1_000_000_000;
small = (b0 << 24)
| ((0xff & buffer[pos + 1]) << 16)
| ((0xff & buffer[pos + 2]) << 8)
| (0xff & buffer[pos + 3]);
small ^= mask;
break;
default:
break;
}
if (small == 0) {
if (frac == 0) {
if (mask != 0) {
longValue = -longValue;
}
compactDecimal = BigDecimal.valueOf(longValue, 0);
} else if (longValue == 0) {
if (scale < 9) {
compactDecimal = DECIMAL_ZERO_1_SCALE;
} else {
compactDecimal = BigDecimal.valueOf(longValue, scale);
}
} else {
if (scale < 9) {
long unscaleValue = longValue * 10;
if (unscaleValue >= longValue) {
if (mask != 0) {
unscaleValue = -unscaleValue;
}
if (unscaleValue == 10) {
compactDecimal = DECIMAL_ONE_1_SCALE;
} else {
compactDecimal = BigDecimal.valueOf(unscaleValue, 1);
}
}
} else {
long unscaleValue = longValue * power;
if (unscaleValue >= longValue) {
if (mask != 0) {
unscaleValue = -unscaleValue;
}
compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
}
}
}
} else {
long unscaleValue = longValue * power + small;
if (unscaleValue >= longValue) {
if (mask != 0) {
unscaleValue = -unscaleValue;
}
compactDecimal = BigDecimal.valueOf(unscaleValue, scale);
}
}
return compactDecimal;
}
/**
* Return big decimal from buffer.
*
@@ -1445,8 +1653,9 @@ public class LogBuffer {
}
d_copy[begin] ^= 0x80; /* restore sign */
String decimal = String.valueOf(buf, 0, pos);
return new BigDecimal(decimal);
// String decimal = String.valueOf(buf, 0, pos);
// return new BigDecimal(decimal);
return new BigDecimal(buf, 0, pos);
}
/**

67
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogDecoder.java
View File

@@ -197,8 +197,9 @@ public final class LogDecoder {
}
case LogEvent.SLAVE_EVENT: /* can never happen (unused event) */
{
if (logger.isWarnEnabled()) logger.warn("Skipping unsupported SLAVE_EVENT from: "
+ context.getLogPosition());
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported SLAVE_EVENT from: " + context.getLogPosition());
}
break;
}
case LogEvent.CREATE_FILE_EVENT: {
@@ -264,22 +265,25 @@ public final class LogDecoder {
return descriptionEvent;
}
case LogEvent.PRE_GA_WRITE_ROWS_EVENT: {
if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: "
+ context.getLogPosition());
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported PRE_GA_WRITE_ROWS_EVENT from: " + context.getLogPosition());
}
// ev = new Write_rows_log_event_old(buf, event_len,
// description_event);
break;
}
case LogEvent.PRE_GA_UPDATE_ROWS_EVENT: {
if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: "
+ context.getLogPosition());
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported PRE_GA_UPDATE_ROWS_EVENT from: " + context.getLogPosition());
}
// ev = new Update_rows_log_event_old(buf, event_len,
// description_event);
break;
}
case LogEvent.PRE_GA_DELETE_ROWS_EVENT: {
if (logger.isWarnEnabled()) logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: "
+ context.getLogPosition());
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported PRE_GA_DELETE_ROWS_EVENT from: " + context.getLogPosition());
}
// ev = new Delete_rows_log_event_old(buf, event_len,
// description_event);
break;
@@ -356,10 +360,16 @@ public final class LogDecoder {
return event;
}
case LogEvent.TRANSACTION_PAYLOAD_EVENT: {
TransactionPayloadLogEvent event = new TransactionPayloadLogEvent(header, buffer, descriptionEvent);
/* updating position in context */
logPosition.position = header.getLogPos();
return event;
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported MySQL TRANSACTION_PAYLOAD_EVENT from: " + context.getLogPosition());
}
break;
// TransactionPayloadLogEvent event = new TransactionPayloadLogEvent(header,
// buffer, descriptionEvent);
// /* updating position in context */
// logPosition.position = header.getLogPos();
// return event;
}
case LogEvent.VIEW_CHANGE_EVENT: {
ViewChangeEvent event = new ViewChangeEvent(header, buffer, descriptionEvent);
@@ -418,6 +428,39 @@ public final class LogDecoder {
logPosition.position = header.getLogPos();
return event;
}
case LogEvent.HEARTBEAT_LOG_EVENT_V2: {
HeartbeatV2LogEvent event = new HeartbeatV2LogEvent(header, buffer, descriptionEvent);
/* updating position in context */
logPosition.position = header.getLogPos();
return event;
}
case LogEvent.QUERY_COMPRESSED_EVENT: {
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported MaraiDB QUERY_COMPRESSED_EVENT from: " + context.getLogPosition());
}
break;
}
case LogEvent.WRITE_ROWS_COMPRESSED_EVENT_V1:
case LogEvent.WRITE_ROWS_COMPRESSED_EVENT: {
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported MaraiDB WRITE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
}
break;
}
case LogEvent.UPDATE_ROWS_COMPRESSED_EVENT_V1:
case LogEvent.UPDATE_ROWS_COMPRESSED_EVENT: {
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported MaraiDB UPDATE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
}
break;
}
case LogEvent.DELETE_ROWS_COMPRESSED_EVENT_V1:
case LogEvent.DELETE_ROWS_COMPRESSED_EVENT: {
if (logger.isWarnEnabled()) {
logger.warn("Skipping unsupported MaraiDB DELETE_ROWS_COMPRESSED_EVENT from: " + context.getLogPosition());
}
break;
}
default:
/*
* Create an object of Ignorable_log_event for unrecognized

23
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/LogEvent.java
View File

@@ -178,7 +178,10 @@ public abstract class LogEvent {
/* mysql 8.0.20 */
public static final int TRANSACTION_PAYLOAD_EVENT = 40;
public static final int MYSQL_ENUM_END_EVENT = 41;
/* mysql 8.0.26 */
public static final int HEARTBEAT_LOG_EVENT_V2 = 41;
public static final int MYSQL_ENUM_END_EVENT = 42;
// mariaDb 5.5.34
/* New MySQL/Sun events are to be added right above this comment */
@@ -210,8 +213,23 @@ public abstract class LogEvent {
public static final int START_ENCRYPTION_EVENT = 164;
// mariadb 10.10.1
/*
* Compressed binlog event. Note that the order between WRITE/UPDATE/DELETE
* events is significant; this is so that we can convert from the compressed to
* the uncompressed event type with (type-WRITE_ROWS_COMPRESSED_EVENT +
* WRITE_ROWS_EVENT) and similar for _V1.
*/
public static final int QUERY_COMPRESSED_EVENT = 165;
public static final int WRITE_ROWS_COMPRESSED_EVENT_V1 = 166;
public static final int UPDATE_ROWS_COMPRESSED_EVENT_V1 = 167;
public static final int DELETE_ROWS_COMPRESSED_EVENT_V1 = 168;
public static final int WRITE_ROWS_COMPRESSED_EVENT = 169;
public static final int UPDATE_ROWS_COMPRESSED_EVENT = 170;
public static final int DELETE_ROWS_COMPRESSED_EVENT = 171;
/** end marker */
public static final int ENUM_END_EVENT = 165;
public static final int ENUM_END_EVENT = 171;
/**
* 1 byte length, 1 byte format Length is total length in bytes, including 2
@@ -358,6 +376,7 @@ public abstract class LogEvent {
case INCIDENT_EVENT:
return "Incident";
case HEARTBEAT_LOG_EVENT:
case HEARTBEAT_LOG_EVENT_V2:
return "Heartbeat";
case IGNORABLE_LOG_EVENT:
return "Ignorable";

1
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/NameCache.java Normal file
View File

File diff suppressed because one or more lines are too long

17
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/GtidLogEvent.java
View File

@@ -18,6 +18,7 @@ public class GtidLogEvent extends LogEvent {
// / Length of SID in event encoding
public static final int ENCODED_SID_LENGTH = 16;
public static final int LOGICAL_TIMESTAMP_TYPE_CODE = 2;
public static final UUID UUID_ZERO = new UUID(0, 0);
private boolean commitFlag;
private UUID sid;
@@ -35,11 +36,13 @@ public class GtidLogEvent extends LogEvent {
buffer.position(commonHeaderLen);
commitFlag = (buffer.getUint8() != 0); // ENCODED_FLAG_LENGTH
byte[] bs = buffer.getData(ENCODED_SID_LENGTH);
ByteBuffer bb = ByteBuffer.wrap(bs);
long high = bb.getLong();
long low = bb.getLong();
sid = new UUID(high, low);
long high = buffer.getBeLong64();
long low = buffer.getBeLong64();
if (high == 0 && low == 0) {
sid = UUID_ZERO;
} else {
sid = new UUID(high, low);
}
gno = buffer.getLong64();
@@ -73,11 +76,11 @@ public class GtidLogEvent extends LogEvent {
return gno;
}
public Long getLastCommitted() {
public long getLastCommitted() {
return lastCommitted;
}
public Long getSequenceNumber() {
public long getSequenceNumber() {
return sequenceNumber;
}

4
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatLogEvent.java
View File

@@ -3,6 +3,8 @@ package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import java.nio.charset.StandardCharsets;
/**
* <pre>
* Replication event to ensure to slave that master is alive.
@@ -35,7 +37,7 @@ public class HeartbeatLogEvent extends LogEvent {
identLen = FN_REFLEN - 1;
}
logIdent = buffer.getFullString(commonHeaderLen, identLen, LogBuffer.ISO_8859_1);
logIdent = buffer.getFullString(commonHeaderLen, identLen, StandardCharsets.ISO_8859_1);
}
public int getIdentLen() {

40
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/HeartbeatV2LogEvent.java Normal file
View File

@@ -0,0 +1,40 @@
package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
/**
* <pre>
* Replication event to ensure to replica that source is alive.
* The event is originated by source's dump thread and sent straight to
* replica without being logged. Slave itself does not store it in relay log
* but rather uses a data for immediate checks and throws away the event.
* Two members of the class m_log_filename and m_log_position comprise
* @see the rpl_event_coordinates instance. The coordinates that a heartbeat
* instance carries correspond to the last event source has sent from
* its binlog.
* Also this event will be generated only for the source server with
* version > 8.0.26
* </pre>
*
* @author jianghang 2022-09-01 下午16:36:29
* @version 1.1.6
* @since mysql 8.0.26
*/
public class HeartbeatV2LogEvent extends LogEvent {
private byte[] payload;
public HeartbeatV2LogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
super(header);
final int commonHeaderLen = descriptionEvent.commonHeaderLen;
int payloadLenth = buffer.limit() - commonHeaderLen;
// see : https://github.com/mysql/mysql-server/commit/59e590738a772b74ad50c4a57c86aaa1bc6501c7#diff-184e9a7d8a58f080974e475d4199fe5c6da5518c8a2811cc5df5988c8f9e9797
payload = buffer.getData(payloadLenth);
}
public byte[] getPayload() {
return payload;
}
}

2
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/IncidentLogEvent.java
View File

@@ -67,7 +67,7 @@ public final class IncidentLogEvent extends LogEvent {
incident = incidentNumber;
buffer.position(commonHeaderLen + postHeaderLen);
message = buffer.getString();
message = buffer.getName();
}
public final int getIncident() {

62
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/QueryLogEvent.java
View File

@@ -433,7 +433,7 @@ public class QueryLogEvent extends LogEvent {
private int serverCollation = -1;
private int tvSec = -1;
private BigInteger ddlXid = BigInteger.valueOf(-1L);
private String charsetName;
private Charset charset;
private String timezone;
@@ -498,12 +498,12 @@ public class QueryLogEvent extends LogEvent {
/* A 2nd variable part; this is common to all versions */
final int queryLen = dataLen - dbLen - 1;
dbname = buffer.getFixString(dbLen + 1);
dbname = buffer.getFixName(dbLen + 1);
if (clientCharset >= 0) {
charsetName = CharsetConversion.getJavaCharset(clientCharset);
charset = CharsetConversion.getNioCharset(clientCharset);
if ((charsetName != null) && (Charset.isSupported(charsetName))) {
query = buffer.getFixString(queryLen, charsetName);
if (charset != null) {
query = buffer.getFixString(queryLen, charset);
} else {
logger.warn("unsupported character set in query log: " + "\n ID = " + clientCharset + ", Charset = "
+ CharsetConversion.getCharset(clientCharset) + ", Collation = "
@@ -603,6 +603,13 @@ public class QueryLogEvent extends LogEvent {
*/
public static final int Q_HRNOW = 128;
/**
* Support MariaDB 10.10.1
*/
public static final int Q_XID = 129;
public static final int Q_GTID_FLAGS3 = 130;
private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
int code = -1;
try {
@@ -613,10 +620,10 @@ public class QueryLogEvent extends LogEvent {
break;
case Q_SQL_MODE_CODE:
sql_mode = buffer.getLong64(); // QQ: Fix when sql_mode
// is ulonglong
// is ulonglong
break;
case Q_CATALOG_NZ_CODE:
catalog = buffer.getString();
catalog = buffer.getName();
break;
case Q_AUTO_INCREMENT:
autoIncrementIncrement = buffer.getUint16();
@@ -632,7 +639,7 @@ public class QueryLogEvent extends LogEvent {
serverCollation = buffer.getUint16();
break;
case Q_TIME_ZONE_CODE:
timezone = buffer.getString();
timezone = buffer.getName();
break;
case Q_CATALOG_CODE: /* for 5.0.x where 0<=x<=3 masters */
final int len = buffer.getUint8();
@@ -656,8 +663,8 @@ public class QueryLogEvent extends LogEvent {
buffer.forward(4);
break;
case Q_INVOKER:
user = buffer.getString();
host = buffer.getString();
user = buffer.getName();
host = buffer.getName();
break;
case Q_MICROSECONDS:
// when.tv_usec= uint3korr(pos);
@@ -678,7 +685,7 @@ public class QueryLogEvent extends LogEvent {
String mtsAccessedDbNames[] = new String[mtsAccessedDbs];
for (int i = 0; i < mtsAccessedDbs && buffer.position() < end; i++) {
int length = end - buffer.position();
mtsAccessedDbNames[i] = buffer.getFixString(length < NAME_LEN ? length : NAME_LEN);
mtsAccessedDbNames[i] = buffer.getFixName(length < NAME_LEN ? length : NAME_LEN);
}
break;
case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
@@ -701,6 +708,25 @@ public class QueryLogEvent extends LogEvent {
// int when_sec_part = buffer.getUint24();
buffer.forward(3);
break;
case Q_XID:
// xid= uint8korr(pos);
buffer.forward(8);
break;
case Q_GTID_FLAGS3:
// gtid_flags_extra= *pos++;
// if (gtid_flags_extra & (Gtid_log_event::FL_COMMIT_ALTER_E1 |
// Gtid_log_event::FL_ROLLBACK_ALTER_E1)) {
// sa_seq_no = uint8korr(pos);
// pos+= 8;
// }
int gtid_flags_extra = buffer.getUint8();
final int FL_COMMIT_ALTER_E1= 4;
final int FL_ROLLBACK_ALTER_E1= 8;
if ((gtid_flags_extra & (FL_COMMIT_ALTER_E1 | FL_ROLLBACK_ALTER_E1))> 0) {
buffer.forward(8);
}
break;
default:
/*
* That's why you must write status vars in growing
@@ -740,16 +766,22 @@ public class QueryLogEvent extends LogEvent {
return "Q_TABLE_MAP_FOR_UPDATE_CODE";
case Q_MASTER_DATA_WRITTEN_CODE:
return "Q_MASTER_DATA_WRITTEN_CODE";
case Q_UPDATED_DB_NAMES:
return "Q_UPDATED_DB_NAMES";
case Q_INVOKER:
return "case Q_INVOKER";
case Q_MICROSECONDS:
return "Q_MICROSECONDS";
case Q_UPDATED_DB_NAMES:
return "Q_UPDATED_DB_NAMES";
case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
return "Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP";
case Q_DDL_LOGGED_WITH_XID:
return "Q_DDL_LOGGED_WITH_XID";
case Q_DEFAULT_COLLATION_FOR_UTF8MB4:
return "Q_DEFAULT_COLLATION_FOR_UTF8MB4";
case Q_SQL_REQUIRE_PRIMARY_KEY:
return "Q_SQL_REQUIRE_PRIMARY_KEY";
case Q_HRNOW:
return "Q_HRNOW";
}
return "CODE#" + code;
}
@@ -797,8 +829,8 @@ public class QueryLogEvent extends LogEvent {
return autoIncrementOffset;
}
public final String getCharsetName() {
return charsetName;
public final Charset getCharset() {
return charset;
}
public final String getTimezone() {

71
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
View File

@@ -1,8 +1,11 @@
package com.taobao.tddl.dbsync.binlog.event;
import java.io.Serializable;
import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.sql.Types;
import java.time.ZoneId;
import java.time.zone.ZoneRules;
import java.util.BitSet;
import org.apache.commons.logging.Log;
@@ -23,7 +26,11 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
*/
public final class RowsLogBuffer {
protected static final Log logger = LogFactory.getLog(RowsLogBuffer.class);
protected static final Log logger = LogFactory.getLog(RowsLogBuffer.class);
public static final Long[] longCache = new Long[1024 * 128];
public static final int longCacheLimit = longCache.length + 127;
public static final Integer[] integerCache = new Integer[1024 * 128];
public static final int integerCacheLimit = longCache.length + 127;
public static final long DATETIMEF_INT_OFS = 0x8000000000L;
public static final long TIMEF_INT_OFS = 0x800000L;
@@ -33,7 +40,7 @@ public final class RowsLogBuffer {
private final LogBuffer buffer;
private final int columnLen;
private final int jsonColumnCount;
private final String charsetName;
private final Charset charset;
private final BitSet nullBits;
private int nullBitIndex;
@@ -47,10 +54,10 @@ public final class RowsLogBuffer {
private int length;
private Serializable value;
public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName, int jsonColumnCount, boolean partial){
public RowsLogBuffer(LogBuffer buffer, final int columnLen, Charset charset, int jsonColumnCount, boolean partial){
this.buffer = buffer;
this.columnLen = columnLen;
this.charsetName = charsetName;
this.charset = charset;
this.partial = partial;
this.jsonColumnCount = jsonColumnCount;
this.nullBits = new BitSet(columnLen);
@@ -309,7 +316,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = unsigned ? Long.valueOf(buffer.getUint32()) :
// Integer.valueOf(buffer.getInt32());
value = Integer.valueOf(buffer.getInt32());
value = valueOf(buffer.getInt32());
javaType = Types.INTEGER;
length = 4;
break;
@@ -318,7 +325,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint8() :
// buffer.getInt8());
value = Integer.valueOf(buffer.getInt8());
value = valueOf(buffer.getInt8());
javaType = Types.TINYINT; // java.sql.Types.INTEGER;
length = 1;
break;
@@ -327,7 +334,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint16() :
// buffer.getInt16());
value = Integer.valueOf((short) buffer.getInt16());
value = valueOf((short) buffer.getInt16());
javaType = Types.SMALLINT; // java.sql.Types.INTEGER;
length = 2;
break;
@@ -336,7 +343,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = Integer.valueOf(unsigned ? buffer.getUint24() :
// buffer.getInt24());
value = Integer.valueOf(buffer.getInt24());
value = valueOf(buffer.getInt24());
javaType = Types.INTEGER;
length = 3;
break;
@@ -345,7 +352,7 @@ public final class RowsLogBuffer {
// XXX: How to check signed / unsigned?
// value = unsigned ? buffer.getUlong64()) :
// Long.valueOf(buffer.getLong64());
value = Long.valueOf(buffer.getLong64());
value = valueOf(buffer.getLong64());
javaType = Types.BIGINT; // Types.INTEGER;
length = 8;
break;
@@ -870,7 +877,7 @@ public final class RowsLogBuffer {
}
// logger.warn("MYSQL_TYPE_ENUM : This enumeration value is "
// + "only used internally and cannot exist in a binlog!");
value = Integer.valueOf(int32);
value = valueOf(int32);
javaType = Types.INTEGER;
length = len;
break;
@@ -1017,7 +1024,7 @@ public final class RowsLogBuffer {
javaType = Types.VARBINARY;
value = binary;
} else {
value = buffer.getFullString(len, charsetName);
value = buffer.getFullString(len, charset);
javaType = Types.VARCHAR;
}
@@ -1039,7 +1046,7 @@ public final class RowsLogBuffer {
javaType = Types.BINARY;
value = binary;
} else {
value = buffer.getFullString(len, charsetName);
value = buffer.getFullString(len, charset);
javaType = Types.CHAR; // Types.VARCHAR;
}
length = len;
@@ -1074,7 +1081,7 @@ public final class RowsLogBuffer {
len,
columnName,
columnIndex,
charsetName);
charset);
value = builder.toString();
buffer.position(position + len);
} else {
@@ -1088,9 +1095,9 @@ public final class RowsLogBuffer {
Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
buffer,
len - 1,
charsetName);
charset);
StringBuilder builder = new StringBuilder();
jsonValue.toJsonString(builder, charsetName);
jsonValue.toJsonString(builder, charset);
value = builder.toString();
buffer.position(position + len);
}
@@ -1211,4 +1218,38 @@ public final class RowsLogBuffer {
builder.append('0').append(digits[d]);
}
}
public static Long valueOf(long value) {
if (value >= -128 & value <= 127) {
// if (l >= -128 && l <= 127) {
return Long.valueOf(value);
}
if (value > 127 && value < longCacheLimit) {
int cacheIndex = (int) value - 127;
Long cacheValue = longCache[cacheIndex];
if (cacheValue == null) {
cacheValue = new Long(value);
longCache[cacheIndex] = cacheValue;
}
return cacheValue;
}
return new Long(value);
}
public static Integer valueOf(int value) {
if (value >= -128 & value <= 127) {
// if (l >= -128 && l <= 127) {
return Integer.valueOf(value);
}
if (value > 127 && value < integerCacheLimit) {
int cacheIndex = (int) value - 127;
Integer cacheValue = integerCache[cacheIndex];
if (cacheValue == null) {
cacheValue = new Integer(value);
integerCache[cacheIndex] = cacheValue;
}
return cacheValue;
}
return new Integer(value);
}
}

5
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
View File

@@ -1,5 +1,6 @@
package com.taobao.tddl.dbsync.binlog.event;
import java.nio.charset.Charset;
import java.util.BitSet;
import com.taobao.tddl.dbsync.binlog.exception.TableIdNotFoundException;
@@ -219,8 +220,8 @@ public abstract class RowsLogEvent extends LogEvent {
return changeColumns;
}
public final RowsLogBuffer getRowsBuf(String charsetName) {
return new RowsLogBuffer(rowsBuf, columnLen, charsetName, jsonColumnCount, partial);
public final RowsLogBuffer getRowsBuf(Charset charset) {
return new RowsLogBuffer(rowsBuf, columnLen, charset, jsonColumnCount, partial);
}
public final int getFlags(final int flags) {

4
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsQueryLogEvent.java
View File

@@ -2,6 +2,8 @@ package com.taobao.tddl.dbsync.binlog.event;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import java.nio.charset.StandardCharsets;
/**
* @author jianghang 2013-4-8 上午12:36:29
* @version 1.0.3
@@ -23,7 +25,7 @@ public class RowsQueryLogEvent extends IgnorableLogEvent {
*/
int offset = commonHeaderLen + postHeaderLen + 1;
int len = buffer.limit() - offset;
rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
}
public String getRowsQuery() {

2
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/StartLogEventV3.java
View File

@@ -36,7 +36,7 @@ public class StartLogEventV3 extends LogEvent {
buffer.position(descriptionEvent.commonHeaderLen);
binlogVersion = buffer.getUint16(); // ST_BINLOG_VER_OFFSET
serverVersion = buffer.getFixString(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
serverVersion = buffer.getFixName(ST_SERVER_VER_LEN); // ST_SERVER_VER_OFFSET
}
public StartLogEventV3(){

4
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
View File

@@ -423,10 +423,10 @@ public final class TableMapLogEvent extends LogEvent {
/* Read the variable part of the event */
buffer.position(commonHeaderLen + postHeaderLen);
dbname = buffer.getString();
dbname = buffer.getName();
buffer.forward(1); /* termination null */
// fixed issue #2714
tblname = buffer.getString();
tblname = buffer.getName();
buffer.forward(1); /* termination null */
// Read column information from buffer

5
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UserVarLogEvent.java
View File

@@ -2,6 +2,7 @@ package com.taobao.tddl.dbsync.binlog.event;
import java.io.IOException;
import java.io.Serializable;
import java.nio.charset.Charset;
import com.taobao.tddl.dbsync.binlog.CharsetConversion;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
@@ -100,8 +101,8 @@ public final class UserVarLogEvent extends LogEvent {
value = buffer.getDecimal(precision, scale); // bin2decimal
break;
case STRING_RESULT:
String charsetName = CharsetConversion.getJavaCharset(charsetNumber);
value = buffer.getFixString(valueLen, charsetName);
Charset charset = CharsetConversion.getNioCharset(charsetNumber);
value = buffer.getFixString(valueLen, charset);
break;
case ROW_RESULT:
// this seems to be banned in MySQL altogether

4
dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/mariadb/AnnotateRowsEvent.java
View File

@@ -5,6 +5,8 @@ import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import com.taobao.tddl.dbsync.binlog.event.IgnorableLogEvent;
import com.taobao.tddl.dbsync.binlog.event.LogHeader;
import java.nio.charset.StandardCharsets;
/**
* mariadb的ANNOTATE_ROWS_EVENT类型
*
@@ -23,7 +25,7 @@ public class AnnotateRowsEvent extends IgnorableLogEvent {
int offset = commonHeaderLen + postHeaderLen;
int len = buffer.limit() - offset;
rowsQuery = buffer.getFullString(offset, len, LogBuffer.ISO_8859_1);
rowsQuery = buffer.getFullString(offset, len, StandardCharsets.ISO_8859_1);
}
public String getRowsQuery() {

2
dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java
View File

@@ -52,7 +52,7 @@ public class BaseLogFetcherTest {
event.getHeader().getLogPos() - event.getHeader().getEventLen(),
event.getTable().getDbName(),
event.getTable().getTableName()));
RowsLogBuffer buffer = event.getRowsBuf(charset.name());
RowsLogBuffer buffer = event.getRowsBuf(charset);
BitSet columns = event.getColumns();
BitSet changeColumns = event.getChangeColumns();
while (buffer.nextOneRow(columns)) {