@@ -394,7 +413,7 @@ public class QueryLogEvent extends LogEvent {
// inspection by the DBA
private final long execTime;
private final int errorCode;
- private final long sessionId; /* thread_id */
+ private final long sessionId; /* thread_id */
/**
* 'flags2' is a second set of flags (on top of those in Log_event), for
@@ -412,6 +431,8 @@ public class QueryLogEvent extends LogEvent {
private int clientCharset = -1;
private int clientCollation = -1;
private int serverCollation = -1;
+ private int tvSec = -1;
+ private BigInteger ddlXid = BigInteger.valueOf(-1L);
private String charsetName;
private String timezone;
@@ -562,6 +583,21 @@ public class QueryLogEvent extends LogEvent {
*/
public static final int Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP = 16;
+ /**
+ * The variable carries xid info of 2pc-aware (recoverable) DDL queries.
+ */
+ public static final int Q_DDL_LOGGED_WITH_XID = 17;
+ /**
+ * This variable stores the default collation for the utf8mb4 character set.
+ * Used to support cross-version replication.
+ */
+ public static final int Q_DEFAULT_COLLATION_FOR_UTF8MB4 = 18;
+
+ /**
+ * Replicate sql_require_primary_key.
+ */
+ public static final int Q_SQL_REQUIRE_PRIMARY_KEY = 19;
+
/**
* FROM MariaDB 5.5.34
*/
@@ -625,7 +661,7 @@ public class QueryLogEvent extends LogEvent {
break;
case Q_MICROSECONDS:
// when.tv_usec= uint3korr(pos);
- buffer.forward(3);
+ tvSec = buffer.getInt24();
break;
case Q_UPDATED_DB_NAMES:
int mtsAccessedDbs = buffer.getUint8();
@@ -646,6 +682,19 @@ public class QueryLogEvent extends LogEvent {
}
break;
case Q_EXPLICIT_DEFAULTS_FOR_TIMESTAMP:
+ // thd->variables.explicit_defaults_for_timestamp
+ buffer.forward(1);
+ break;
+ case Q_DDL_LOGGED_WITH_XID:
+ ddlXid = buffer.getUlong64();
+ break;
+ case Q_DEFAULT_COLLATION_FOR_UTF8MB4:
+ // int2store(start,
+ // default_collation_for_utf8mb4_number);
+ buffer.forward(2);
+ break;
+ case Q_SQL_REQUIRE_PRIMARY_KEY:
+ // *start++ = thd->variables.sql_require_primary_key;
buffer.forward(1);
break;
case Q_HRNOW:
@@ -657,8 +706,10 @@ public class QueryLogEvent extends LogEvent {
* That's why you must write status vars in growing
* order of code
*/
- if (logger.isDebugEnabled()) logger.debug("Query_log_event has unknown status vars (first has code: "
- + code + "), skipping the rest of them");
+ if (logger.isDebugEnabled()) {
+ logger.debug("Query_log_event has unknown status vars (first has code: " + code
+ + "), skipping the rest of them");
+ }
break; // Break loop
}
}
@@ -695,6 +746,12 @@ public class QueryLogEvent extends LogEvent {
return "Q_UPDATED_DB_NAMES";
case Q_MICROSECONDS:
return "Q_MICROSECONDS";
+ 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";
}
return "CODE#" + code;
}
@@ -777,6 +834,14 @@ public class QueryLogEvent extends LogEvent {
return serverCollation;
}
+ public int getTvSec() {
+ return tvSec;
+ }
+
+ public BigInteger getDdlXid() {
+ return ddlXid;
+ }
+
/**
* Returns the sql_mode value.
*
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
index 8cd57332..013494d5 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java
@@ -10,6 +10,7 @@ import org.apache.commons.logging.LogFactory;
import com.taobao.tddl.dbsync.binlog.JsonConversion;
import com.taobao.tddl.dbsync.binlog.JsonConversion.Json_Value;
+import com.taobao.tddl.dbsync.binlog.JsonDiffConversion;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -31,22 +32,33 @@ public final class RowsLogBuffer {
private final LogBuffer buffer;
private final int columnLen;
+ private final int jsonColumnCount;
private final String charsetName;
- // private Calendar cal;
private final BitSet nullBits;
private int nullBitIndex;
+ // Read value_options if this is AI for PARTIAL_UPDATE_ROWS_EVENT
+ private final boolean partial;
+ private final BitSet partialBits;
+
private boolean fNull;
private int javaType;
private int length;
private Serializable value;
- public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName){
+ public RowsLogBuffer(LogBuffer buffer, final int columnLen, String charsetName, int jsonColumnCount, boolean partial){
this.buffer = buffer;
this.columnLen = columnLen;
this.charsetName = charsetName;
+ this.partial = partial;
+ this.jsonColumnCount = jsonColumnCount;
this.nullBits = new BitSet(columnLen);
+ this.partialBits = new BitSet(1);
+ }
+
+ public final boolean nextOneRow(BitSet columns) {
+ return nextOneRow(columns, false);
}
/**
@@ -55,18 +67,30 @@ public final class RowsLogBuffer {
* @see mysql-5.1.60/sql/log_event.cc -
* Rows_log_event::print_verbose_one_row
*/
- public final boolean nextOneRow(BitSet columns) {
+ public final boolean nextOneRow(BitSet columns, boolean after) {
final boolean hasOneRow = buffer.hasRemaining();
if (hasOneRow) {
int column = 0;
for (int i = 0; i < columnLen; i++)
- if (columns.get(i)) column++;
+ if (columns.get(i)) {
+ column++;
+ }
+ if (after && partial) {
+ partialBits.clear();
+ long valueOptions = buffer.getPackedLong();
+ int PARTIAL_JSON_UPDATES = 1;
+ if ((valueOptions & PARTIAL_JSON_UPDATES) != 0) {
+ partialBits.set(1);
+ buffer.forward((jsonColumnCount + 7) / 8);
+ }
+ }
nullBitIndex = 0;
nullBits.clear();
buffer.fillBitmap(nullBits, column);
+
}
return hasOneRow;
}
@@ -77,8 +101,8 @@ public final class RowsLogBuffer {
* @see mysql-5.1.60/sql/log_event.cc -
* Rows_log_event::print_verbose_one_row
*/
- public final Serializable nextValue(final int type, final int meta) {
- return nextValue(type, meta, false);
+ public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta) {
+ return nextValue(columName, columnIndex, type, meta, false);
}
/**
@@ -87,7 +111,8 @@ public final class RowsLogBuffer {
* @see mysql-5.1.60/sql/log_event.cc -
* Rows_log_event::print_verbose_one_row
*/
- public final Serializable nextValue(final int type, final int meta, boolean isBinary) {
+ public final Serializable nextValue(final String columName, final int columnIndex, final int type, final int meta,
+ boolean isBinary) {
fNull = nullBits.get(nullBitIndex++);
if (fNull) {
@@ -97,7 +122,7 @@ public final class RowsLogBuffer {
return null;
} else {
// Extracting field value from packed buffer.
- return fetchValue(type, meta, isBinary);
+ return fetchValue(columName, columnIndex, type, meta, isBinary);
}
}
@@ -248,7 +273,7 @@ public final class RowsLogBuffer {
*
* @see mysql-5.1.60/sql/log_event.cc - log_event_print_value
*/
- final Serializable fetchValue(int type, final int meta, boolean isBinary) {
+ final Serializable fetchValue(String columnName, int columnIndex, int type, final int meta, boolean isBinary) {
int len = 0;
if (type == LogEvent.MYSQL_TYPE_STRING) {
@@ -610,7 +635,7 @@ public final class RowsLogBuffer {
// (u32 % 10000) / 100,
// u32 % 100);
- StringBuilder builder = new StringBuilder(12);
+ StringBuilder builder = new StringBuilder(17);
if (i32 < 0) {
builder.append('-');
}
@@ -1041,17 +1066,34 @@ public final class RowsLogBuffer {
default:
throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta);
}
- if (0 == len) {
- // fixed issue #1 by lava, json column of zero length has no
- // value, value parsing should be skipped
- value = "";
- } else {
+
+ if (partialBits.get(1)) {
+ // print_json_diff
int position = buffer.position();
- Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName);
- StringBuilder builder = new StringBuilder();
- jsonValue.toJsonString(builder, charsetName);
+ StringBuilder builder = JsonDiffConversion.print_json_diff(buffer,
+ len,
+ columnName,
+ columnIndex,
+ charsetName);
value = builder.toString();
buffer.position(position + len);
+ } else {
+ if (0 == len) {
+ // fixed issue #1 by lava, json column of zero length
+ // has no
+ // value, value parsing should be skipped
+ value = "";
+ } else {
+ int position = buffer.position();
+ Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(),
+ buffer,
+ len - 1,
+ charsetName);
+ StringBuilder builder = new StringBuilder();
+ jsonValue.toJsonString(builder, charsetName);
+ value = builder.toString();
+ buffer.position(position + len);
+ }
}
javaType = Types.VARCHAR;
length = len;
@@ -1120,7 +1162,7 @@ public final class RowsLogBuffer {
return length;
}
- private String usecondsToStr(int frac, int meta) {
+ public static String usecondsToStr(int frac, int meta) {
String sec = String.valueOf(frac);
if (meta > 6) {
throw new IllegalArgumentException("unknow useconds meta : " + meta);
@@ -1139,7 +1181,7 @@ public final class RowsLogBuffer {
return sec.substring(0, meta);
}
- private void appendNumber4(StringBuilder builder, int d) {
+ public static void appendNumber4(StringBuilder builder, int d) {
if (d >= 1000) {
builder.append(digits[d / 1000])
.append(digits[(d / 100) % 10])
@@ -1151,7 +1193,7 @@ public final class RowsLogBuffer {
}
}
- private void appendNumber3(StringBuilder builder, int d) {
+ public static void appendNumber3(StringBuilder builder, int d) {
if (d >= 100) {
builder.append(digits[d / 100]).append(digits[(d / 10) % 10]).append(digits[d % 10]);
} else {
@@ -1160,7 +1202,7 @@ public final class RowsLogBuffer {
}
}
- private void appendNumber2(StringBuilder builder, int d) {
+ public static void appendNumber2(StringBuilder builder, int d) {
if (d >= 10) {
builder.append(digits[(d / 10) % 10]).append(digits[d % 10]);
} else {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
index f2ce79b0..f11b0734 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogEvent.java
@@ -5,6 +5,7 @@ import java.util.BitSet;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogEvent;
+import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
/**
* Common base class for all row-containing log events.
@@ -61,6 +62,7 @@ public abstract class RowsLogEvent extends LogEvent {
/** Bitmap denoting columns available */
protected final int columnLen;
+ protected final boolean partial;
protected final BitSet columns;
/**
@@ -71,6 +73,8 @@ public abstract class RowsLogEvent extends LogEvent {
*/
protected final BitSet changeColumns;
+ protected int jsonColumnCount = 0;
+
/** XXX: Don't handle buffer in another thread. */
private final LogBuffer rowsBuf; /*
* The rows in
@@ -109,6 +113,10 @@ public abstract class RowsLogEvent extends LogEvent {
public static final int RW_V_EXTRAINFO_TAG = 0;
public RowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
+ this(header, buffer, descriptionEvent, false);
+ }
+
+ public RowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent, boolean partial){
super(header);
final int commonHeaderLen = descriptionEvent.commonHeaderLen;
@@ -153,9 +161,11 @@ public abstract class RowsLogEvent extends LogEvent {
buffer.position(commonHeaderLen + postHeaderLen + headerLen);
columnLen = (int) buffer.getPackedLong();
+ this.partial = partial;
columns = buffer.getBitmap(columnLen);
- if (header.type == UPDATE_ROWS_EVENT_V1 || header.type == UPDATE_ROWS_EVENT) {
+ if (header.type == UPDATE_ROWS_EVENT_V1 || header.type == UPDATE_ROWS_EVENT
+ || header.type == PARTIAL_UPDATE_ROWS_EVENT) {
changeColumns = buffer.getBitmap(columnLen);
} else {
changeColumns = columns;
@@ -175,6 +185,17 @@ public abstract class RowsLogEvent extends LogEvent {
// delete original table map events stored in the map).
context.clearAllTables();
}
+
+ int jsonColumnCount = 0;
+ int columnCnt = table.getColumnCnt();
+ ColumnInfo[] columnInfo = table.getColumnInfo();
+ for (int i = 0; i < columnCnt; i++) {
+ ColumnInfo info = columnInfo[i];
+ if (info.type == LogEvent.MYSQL_TYPE_JSON) {
+ jsonColumnCount++;
+ }
+ }
+ this.jsonColumnCount = jsonColumnCount;
}
public final long getTableId() {
@@ -194,7 +215,7 @@ public abstract class RowsLogEvent extends LogEvent {
}
public final RowsLogBuffer getRowsBuf(String charsetName) {
- return new RowsLogBuffer(rowsBuf, columnLen, charsetName);
+ return new RowsLogBuffer(rowsBuf, columnLen, charsetName, jsonColumnCount, partial);
}
public final int getFlags(final int flags) {
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
index e6550b27..18201ea9 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/TableMapLogEvent.java
@@ -1,6 +1,8 @@
package com.taobao.tddl.dbsync.binlog.event;
+import java.util.ArrayList;
import java.util.BitSet;
+import java.util.List;
import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogEvent;
@@ -88,6 +90,17 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
* first byte, the ninth is in the least significant bit of the second byte, and
* so on.
*
+ *
+ * | optional metadata fields |
+ * optional metadata fields are stored in Type, Length, Value(TLV) format.
+ * Type takes 1 byte. Length is a packed integer value. Values takes Length
+ * bytes. |
+ * There are some optional metadata defined. They are listed in the table
+ *
+ * @ref Table_table_map_event_optional_metadata. Optional metadata fields follow
+ * null_bits. Whether binlogging an optional metadata is decided by the server.
+ * The order is not defined, so they can be binlogged in any order. |
+ *
*
* The table below lists all column types, along with the numerical identifier
* for it and the size and interpretation of meta-data used to describe the
@@ -284,7 +297,6 @@ import com.taobao.tddl.dbsync.binlog.LogEvent;
* of the geometry: 1, 2, 3, or 4.
*
*
- *
* @author Changyuan.lh
* @version 1.0
*/
@@ -322,20 +334,63 @@ public final class TableMapLogEvent extends LogEvent {
*/
public static final class ColumnInfo {
- public int type;
- public int meta;
+ public int type;
+ public int meta;
+ public String name;
+ public boolean unsigned;
+ public boolean pk;
+ public List set_enum_values;
+ public int charset; // 可以通过CharsetUtil进行转化
+ public int geoType;
+ public boolean nullable;
+
+ @Override
+ public String toString() {
+ return "ColumnInfo [type=" + type + ", meta=" + meta + ", name=" + name + ", unsigned=" + unsigned
+ + ", pk=" + pk + ", set_enum_values=" + set_enum_values + ", charset=" + charset + ", geoType="
+ + geoType + ", nullable=" + nullable + "]";
+ }
}
protected final int columnCnt;
- protected final ColumnInfo[] columnInfo; // buffer for field
- // metadata
+ protected final ColumnInfo[] columnInfo; // buffer
+ // for
+ // field
+ // metadata
protected final long tableId;
protected BitSet nullBits;
/** TM = "Table Map" */
- public static final int TM_MAPID_OFFSET = 0;
- public static final int TM_FLAGS_OFFSET = 6;
+ public static final int TM_MAPID_OFFSET = 0;
+ public static final int TM_FLAGS_OFFSET = 6;
+
+ // UNSIGNED flag of numeric columns
+ public static final int SIGNEDNESS = 1;
+ // Default character set of string columns
+ public static final int DEFAULT_CHARSET = 2;
+ // Character set of string columns
+ public static final int COLUMN_CHARSET = 3;
+ public static final int COLUMN_NAME = 4;
+ // String value of SET columns
+ public static final int SET_STR_VALUE = 5;
+ // String value of ENUM columns
+ public static final int ENUM_STR_VALUE = 6;
+ // Real type of geometry columns
+ public static final int GEOMETRY_TYPE = 7;
+ // Primary key without prefix
+ public static final int SIMPLE_PRIMARY_KEY = 8;
+ // Primary key with prefix
+ public static final int PRIMARY_KEY_WITH_PREFIX = 9;
+
+ private int default_charset;
+ private boolean existOptionalMetaData = false;
+
+ private static final class Pair {
+
+ public int col_index;
+ public int col_charset;
+ }
/**
* Constructor used by slave to read the event from the binary log.
@@ -379,7 +434,91 @@ public final class TableMapLogEvent extends LogEvent {
final int fieldSize = (int) buffer.getPackedLong();
decodeFields(buffer, fieldSize);
nullBits = buffer.getBitmap(columnCnt);
+
+ for (int i = 0; i < columnCnt; i++) {
+ if (nullBits.get(i)) {
+ columnInfo[i].nullable = true;
+ }
+ }
+ /*
+ * After null_bits field, there are some new fields for extra
+ * metadata.
+ */
+ existOptionalMetaData = false;
+ List defaultCharsetPairs = null;
+ List columnCharsets = null;
+ while (buffer.hasRemaining()) {
+ // optional metadata fields
+ int type = buffer.getUint8();
+ int len = (int) buffer.getPackedLong();
+
+ switch (type) {
+ case SIGNEDNESS:
+ parse_signedness(buffer, len);
+ break;
+ case DEFAULT_CHARSET:
+ defaultCharsetPairs = parse_default_charset(buffer, len);
+ break;
+ case COLUMN_CHARSET:
+ columnCharsets = parse_column_charset(buffer, len);
+ break;
+ case COLUMN_NAME:
+ // set @@global.binlog_row_metadata='FULL'
+ // 主要是补充列名相关信息
+ existOptionalMetaData = true;
+ parse_column_name(buffer, len);
+ break;
+ case SET_STR_VALUE:
+ parse_set_str_value(buffer, len, true);
+ break;
+ case ENUM_STR_VALUE:
+ parse_set_str_value(buffer, len, false);
+ break;
+ case GEOMETRY_TYPE:
+ parse_geometry_type(buffer, len);
+ break;
+ case SIMPLE_PRIMARY_KEY:
+ parse_simple_pk(buffer, len);
+ break;
+ case PRIMARY_KEY_WITH_PREFIX:
+ parse_pk_with_prefix(buffer, len);
+ break;
+ default:
+ throw new IllegalArgumentException("unknow type : " + type);
+ }
+ }
+
+ if (existOptionalMetaData) {
+ int index = 0;
+ int char_col_index = 0;
+ for (int i = 0; i < columnCnt; i++) {
+ int cs = -1;
+ int type = getRealType(columnInfo[i].type, columnInfo[i].meta);
+ if (is_character_type(type)) {
+ if (defaultCharsetPairs != null && !defaultCharsetPairs.isEmpty()) {
+ if (index < defaultCharsetPairs.size()
+ && char_col_index == defaultCharsetPairs.get(index).col_index) {
+ cs = defaultCharsetPairs.get(index).col_charset;
+ index++;
+ } else {
+ cs = default_charset;
+ }
+
+ char_col_index++;
+ } else if (columnCharsets != null) {
+ cs = columnCharsets.get(index);
+ index++;
+ }
+
+ columnInfo[i].charset = cs;
+ }
+ }
+ }
}
+
+ // for (int i = 0; i < columnCnt; i++) {
+ // System.out.println(columnInfo[i]);
+ // }
}
/**
@@ -459,6 +598,192 @@ public final class TableMapLogEvent extends LogEvent {
buffer.limit(limit);
}
+ private void parse_signedness(LogBuffer buffer, int length) {
+ // stores the signedness flags extracted from field
+ List datas = new ArrayList();
+ for (int i = 0; i < length; i++) {
+ int ut = buffer.getUint8();
+ for (int c = 0x80; c != 0; c >>= 1) {
+ datas.add((ut & c) > 0);
+ }
+ }
+
+ int index = 0;
+ for (int i = 0; i < columnCnt; i++) {
+ if (is_numeric_type(columnInfo[i].type)) {
+ columnInfo[i].unsigned = datas.get(index);
+ index++;
+ }
+ }
+ }
+
+ private List parse_default_charset(LogBuffer buffer, int length) {
+ // stores collation numbers extracted from field.
+ int limit = buffer.position() + length;
+ this.default_charset = (int) buffer.getPackedLong();
+ List datas = new ArrayList();
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int col_index = (int) buffer.getPackedLong();
+ int col_charset = (int) buffer.getPackedLong();
+
+ Pair pair = new Pair();
+ pair.col_index = col_index;
+ pair.col_charset = col_charset;
+ datas.add(pair);
+ }
+
+ return datas;
+ }
+
+ private List parse_column_charset(LogBuffer buffer, int length) {
+ // stores collation numbers extracted from field.
+ int limit = buffer.position() + length;
+ List datas = new ArrayList();
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int col_charset = (int) buffer.getPackedLong();
+ datas.add(col_charset);
+ }
+
+ return datas;
+ }
+
+ private void parse_column_name(LogBuffer buffer, int length) {
+ // stores column names extracted from field
+ int limit = buffer.position() + length;
+ int index = 0;
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int len = (int) buffer.getPackedLong();
+ columnInfo[index++].name = buffer.getFixString(len);
+ }
+ }
+
+ private void parse_set_str_value(LogBuffer buffer, int length, boolean set) {
+ // stores SET/ENUM column's string values extracted from
+ // field. Each SET/ENUM column's string values are stored
+ // into a string separate vector. All of them are stored
+ // in 'vec'.
+ int limit = buffer.position() + length;
+ List> datas = new ArrayList>();
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int count = (int) buffer.getPackedLong();
+ List data = new ArrayList(count);
+ for (int i = 0; i < count; i++) {
+ int len1 = (int) buffer.getPackedLong();
+ data.add(buffer.getFixString(len1));
+ }
+
+ datas.add(data);
+ }
+
+ int index = 0;
+ for (int i = 0; i < columnCnt; i++) {
+ if (set && getRealType(columnInfo[i].type, columnInfo[i].meta) == LogEvent.MYSQL_TYPE_SET) {
+ columnInfo[i].set_enum_values = datas.get(index);
+ index++;
+ }
+
+ if (!set && getRealType(columnInfo[i].type, columnInfo[i].meta) == LogEvent.MYSQL_TYPE_ENUM) {
+ columnInfo[i].set_enum_values = datas.get(index);
+ index++;
+ }
+ }
+ }
+
+ private void parse_geometry_type(LogBuffer buffer, int length) {
+ // stores geometry column's types extracted from field.
+ int limit = buffer.position() + length;
+
+ List datas = new ArrayList();
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int col_type = (int) buffer.getPackedLong();
+ datas.add(col_type);
+ }
+
+ int index = 0;
+ for (int i = 0; i < columnCnt; i++) {
+ if (columnInfo[i].type == LogEvent.MYSQL_TYPE_GEOMETRY) {
+ columnInfo[i].geoType = datas.get(index);
+ index++;
+ }
+ }
+ }
+
+ private void parse_simple_pk(LogBuffer buffer, int length) {
+ // stores primary key's column information extracted from
+ // field. Each column has an index and a prefix which are
+ // stored as a unit_pair. prefix is always 0 for
+ // SIMPLE_PRIMARY_KEY field.
+
+ int limit = buffer.position() + length;
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int col_index = (int) buffer.getPackedLong();
+ columnInfo[col_index].pk = true;
+ }
+ }
+
+ private void parse_pk_with_prefix(LogBuffer buffer, int length) {
+ // stores primary key's column information extracted from
+ // field. Each column has an index and a prefix which are
+ // stored as a unit_pair.
+ int limit = buffer.position() + length;
+ while (buffer.hasRemaining() && buffer.position() < limit) {
+ int col_index = (int) buffer.getPackedLong();
+ // prefix length, 比如 char(32)
+ @SuppressWarnings("unused")
+ int col_prefix = (int) buffer.getPackedLong();
+ columnInfo[col_index].pk = true;
+ }
+ }
+
+ private boolean is_numeric_type(int type) {
+ switch (type) {
+ case MYSQL_TYPE_TINY:
+ case MYSQL_TYPE_SHORT:
+ case MYSQL_TYPE_INT24:
+ case MYSQL_TYPE_LONG:
+ case MYSQL_TYPE_LONGLONG:
+ case MYSQL_TYPE_NEWDECIMAL:
+ case MYSQL_TYPE_FLOAT:
+ case MYSQL_TYPE_DOUBLE:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private boolean is_character_type(int type) {
+ switch (type) {
+ case MYSQL_TYPE_STRING:
+ case MYSQL_TYPE_VAR_STRING:
+ case MYSQL_TYPE_VARCHAR:
+ case MYSQL_TYPE_BLOB:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ private int getRealType(int type, int meta) {
+ if (type == LogEvent.MYSQL_TYPE_STRING) {
+ if (meta >= 256) {
+ int byte0 = meta >> 8;
+ if ((byte0 & 0x30) != 0x30) {
+ /* a long CHAR() field: see #37426 */
+ type = byte0 | 0x30;
+ } else {
+ switch (byte0) {
+ case LogEvent.MYSQL_TYPE_SET:
+ case LogEvent.MYSQL_TYPE_ENUM:
+ case LogEvent.MYSQL_TYPE_STRING:
+ type = byte0;
+ }
+ }
+ }
+ }
+
+ return type;
+ }
+
public final String getDbName() {
return dbname;
}
@@ -478,4 +803,13 @@ public final class TableMapLogEvent extends LogEvent {
public final long getTableId() {
return tableId;
}
+
+ public boolean isExistOptionalMetaData() {
+ return existOptionalMetaData;
+ }
+
+ public void setExistOptionalMetaData(boolean existOptional) {
+ this.existOptionalMetaData = existOptional;
+ }
+
}
diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java
index f5e5866e..b604e04d 100644
--- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java
+++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/UpdateRowsLogEvent.java
@@ -14,6 +14,11 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer;
public final class UpdateRowsLogEvent extends RowsLogEvent {
public UpdateRowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent){
- super(header, buffer, descriptionEvent);
+ super(header, buffer, descriptionEvent, false);
+ }
+
+ public UpdateRowsLogEvent(LogHeader header, LogBuffer buffer, FormatDescriptionLogEvent descriptionEvent,
+ boolean partial){
+ super(header, buffer, descriptionEvent, partial);
}
}
diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java
index d0c6746d..27ecf3db 100644
--- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java
+++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/BaseLogFetcherTest.java
@@ -68,7 +68,7 @@ public class BaseLogFetcherTest {
// update需要处理before/after
System.out.println("-------> before");
parseOneRow(event, buffer, columns, false);
- if (!buffer.nextOneRow(changeColumns)) {
+ if (!buffer.nextOneRow(changeColumns, true)) {
break;
}
System.out.println("-------> after");
@@ -97,7 +97,7 @@ public class BaseLogFetcherTest {
}
ColumnInfo info = columnInfo[i];
- buffer.nextValue(info.type, info.meta);
+ buffer.nextValue(null , i ,info.type, info.meta);
if (buffer.isNull()) {
//
diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java
index c48082df..3648da68 100644
--- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java
+++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/DirectLogFetcherTest.java
@@ -45,6 +45,7 @@ public class DirectLogFetcherTest extends BaseLogFetcherTest {
parseRowsEvent((WriteRowsLogEvent) event);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
parseRowsEvent((UpdateRowsLogEvent) event);
break;
diff --git a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java
index 7673f71b..aada3de8 100644
--- a/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java
+++ b/dbsync/src/test/java/com/taobao/tddl/dbsync/binlog/FileLogFetcherTest.java
@@ -53,6 +53,7 @@ public class FileLogFetcherTest extends BaseLogFetcherTest {
parseRowsEvent((WriteRowsLogEvent) event);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
parseRowsEvent((UpdateRowsLogEvent) event);
break;
diff --git a/deployer/src/main/resources/mq.yml b/deployer/src/main/resources/mq.yml
index 019d2123..1797bd9b 100644
--- a/deployer/src/main/resources/mq.yml
+++ b/deployer/src/main/resources/mq.yml
@@ -1,4 +1,4 @@
-servers: localhost:9876 #for rocketmq: means the nameserver
+servers: slave1:6667 #for rocketmq: means the nameserver
retries: 0
batchSize: 16384
lingerMs: 1
@@ -13,7 +13,7 @@ flatMessage: true
canalDestinations:
- canalDestination: example
topic: example
- partition: 1
+ partition:
# #对应topic分区数量
# partitionsNum: 3
# partitionHash:
diff --git a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml
index beb61236..c05b5200 100644
--- a/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml
+++ b/deployer/src/main/resources/spring/tsdb/h2-tsdb.xml
@@ -23,7 +23,7 @@
-
+
diff --git a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml
index 58cab638..87f407b3 100644
--- a/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml
+++ b/deployer/src/main/resources/spring/tsdb/mysql-tsdb.xml
@@ -23,7 +23,7 @@
-
+
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
index 8de2acfd..dfa2783f 100644
--- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/BioSocketChannel.java
@@ -88,8 +88,9 @@ public class BioSocketChannel implements SocketChannel {
}
}
if (remain > 0 && accTimeout >= timeout) {
- throw new SocketTimeoutException("Timeout occurred, failed to read " + readSize + " bytes in " + timeout
- + " milliseconds.");
+ throw new SocketTimeoutException("Timeout occurred, failed to read total " + readSize + " bytes in "
+ + timeout + " milliseconds, actual read only " + (readSize - remain)
+ + " bytes");
}
return data;
}
@@ -120,8 +121,8 @@ public class BioSocketChannel implements SocketChannel {
}
if (n < len && accTimeout >= timeout) {
- throw new SocketTimeoutException("Timeout occurred, failed to read " + len + " bytes in " + timeout
- + " milliseconds.");
+ throw new SocketTimeoutException("Timeout occurred, failed to read total " + len + " bytes in " + timeout
+ + " milliseconds, actual read only " + n + " bytes");
}
}
diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java
new file mode 100644
index 00000000..ba98a87f
--- /dev/null
+++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/utils/CharsetUtil.java
@@ -0,0 +1,258 @@
+package com.alibaba.otter.canal.parse.driver.mysql.utils;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.commons.lang.StringUtils;
+
+/**
+ * mysql collation转换mapping关系表
+ *
+ * @author agapple 2018年11月5日 下午1:01:15
+ * @since 1.1.2
+ */
+public class CharsetUtil {
+
+ private static final String[] INDEX_TO_CHARSET = new String[2048];
+ private static final Map CHARSET_TO_INDEX = new HashMap();
+ static {
+ INDEX_TO_CHARSET[1] = "big5";
+ INDEX_TO_CHARSET[84] = "big5";
+
+ INDEX_TO_CHARSET[3] = "dec8";
+ INDEX_TO_CHARSET[69] = "dec8";
+
+ INDEX_TO_CHARSET[4] = "cp850";
+ INDEX_TO_CHARSET[80] = "cp850";
+
+ INDEX_TO_CHARSET[6] = "hp8";
+ INDEX_TO_CHARSET[72] = "hp8";
+
+ INDEX_TO_CHARSET[7] = "koi8r";
+ INDEX_TO_CHARSET[74] = "koi8r";
+
+ INDEX_TO_CHARSET[5] = "latin1";
+ INDEX_TO_CHARSET[8] = "latin1";
+ INDEX_TO_CHARSET[15] = "latin1";
+ INDEX_TO_CHARSET[31] = "latin1";
+ INDEX_TO_CHARSET[47] = "latin1";
+ INDEX_TO_CHARSET[48] = "latin1";
+ INDEX_TO_CHARSET[49] = "latin1";
+ INDEX_TO_CHARSET[94] = "latin1";
+
+ INDEX_TO_CHARSET[9] = "latin2";
+ INDEX_TO_CHARSET[21] = "latin2";
+ INDEX_TO_CHARSET[27] = "latin2";
+ INDEX_TO_CHARSET[77] = "latin2";
+
+ INDEX_TO_CHARSET[10] = "swe7";
+ INDEX_TO_CHARSET[82] = "swe7";
+
+ INDEX_TO_CHARSET[11] = "ascii";
+ INDEX_TO_CHARSET[65] = "ascii";
+
+ INDEX_TO_CHARSET[12] = "ujis";
+ INDEX_TO_CHARSET[91] = "ujis";
+
+ INDEX_TO_CHARSET[13] = "sjis";
+ INDEX_TO_CHARSET[88] = "sjis";
+
+ INDEX_TO_CHARSET[16] = "hebrew";
+ INDEX_TO_CHARSET[71] = "hebrew";
+
+ INDEX_TO_CHARSET[18] = "tis620";
+ INDEX_TO_CHARSET[69] = "tis620";
+
+ INDEX_TO_CHARSET[19] = "euckr";
+ INDEX_TO_CHARSET[85] = "euckr";
+
+ INDEX_TO_CHARSET[22] = "koi8u";
+ INDEX_TO_CHARSET[75] = "koi8u";
+
+ INDEX_TO_CHARSET[24] = "gb2312";
+ INDEX_TO_CHARSET[86] = "gb2312";
+
+ INDEX_TO_CHARSET[25] = "greek";
+ INDEX_TO_CHARSET[70] = "greek";
+
+ INDEX_TO_CHARSET[26] = "cp1250";
+ INDEX_TO_CHARSET[34] = "cp1250";
+ INDEX_TO_CHARSET[44] = "cp1250";
+ INDEX_TO_CHARSET[66] = "cp1250";
+ INDEX_TO_CHARSET[99] = "cp1250";
+
+ INDEX_TO_CHARSET[28] = "gbk";
+ INDEX_TO_CHARSET[87] = "gbk";
+
+ INDEX_TO_CHARSET[30] = "latin5";
+ INDEX_TO_CHARSET[78] = "latin5";
+
+ INDEX_TO_CHARSET[32] = "armscii8";
+ INDEX_TO_CHARSET[64] = "armscii8";
+
+ INDEX_TO_CHARSET[33] = "utf8";
+ INDEX_TO_CHARSET[83] = "utf8";
+ for (int i = 192; i <= 223; i++) {
+ INDEX_TO_CHARSET[i] = "utf8";
+ }
+ for (int i = 336; i <= 337; i++) {
+ INDEX_TO_CHARSET[i] = "utf8";
+ }
+ for (int i = 352; i <= 357; i++) {
+ INDEX_TO_CHARSET[i] = "utf8";
+ }
+ INDEX_TO_CHARSET[368] = "utf8";
+ INDEX_TO_CHARSET[2047] = "utf8";
+
+ INDEX_TO_CHARSET[35] = "ucs2";
+ INDEX_TO_CHARSET[90] = "ucs2";
+ for (int i = 128; i <= 151; i++) {
+ INDEX_TO_CHARSET[i] = "ucs2";
+ }
+ INDEX_TO_CHARSET[159] = "ucs2";
+ for (int i = 358; i <= 360; i++) {
+ INDEX_TO_CHARSET[i] = "ucs2";
+ }
+
+ INDEX_TO_CHARSET[36] = "cp866";
+ INDEX_TO_CHARSET[68] = "cp866";
+
+ INDEX_TO_CHARSET[37] = "keybcs2";
+ INDEX_TO_CHARSET[73] = "keybcs2";
+
+ INDEX_TO_CHARSET[38] = "macce";
+ INDEX_TO_CHARSET[43] = "macce";
+
+ INDEX_TO_CHARSET[39] = "macroman";
+ INDEX_TO_CHARSET[53] = "macroman";
+
+ INDEX_TO_CHARSET[40] = "cp852";
+ INDEX_TO_CHARSET[81] = "cp852";
+
+ INDEX_TO_CHARSET[20] = "latin7";
+ INDEX_TO_CHARSET[41] = "latin7";
+ INDEX_TO_CHARSET[42] = "latin7";
+ INDEX_TO_CHARSET[79] = "latin7";
+
+ INDEX_TO_CHARSET[45] = "utf8mb4";
+ INDEX_TO_CHARSET[46] = "utf8mb4";
+ for (int i = 224; i <= 247; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ for (int i = 255; i <= 271; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ for (int i = 273; i <= 275; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ for (int i = 277; i <= 294; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ for (int i = 296; i <= 298; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ INDEX_TO_CHARSET[300] = "utf8mb4";
+ for (int i = 303; i <= 307; i++) {
+ INDEX_TO_CHARSET[i] = "utf8mb4";
+ }
+ INDEX_TO_CHARSET[326] = "utf8mb4";
+ INDEX_TO_CHARSET[328] = "utf8mb4";
+
+ INDEX_TO_CHARSET[14] = "cp1251";
+ INDEX_TO_CHARSET[23] = "cp1251";
+ INDEX_TO_CHARSET[50] = "cp1251";
+ INDEX_TO_CHARSET[51] = "cp1251";
+ INDEX_TO_CHARSET[52] = "cp1251";
+
+ INDEX_TO_CHARSET[54] = "utf16";
+ INDEX_TO_CHARSET[55] = "utf16";
+ for (int i = 101; i <= 124; i++) {
+ INDEX_TO_CHARSET[i] = "utf16";
+ }
+ INDEX_TO_CHARSET[327] = "utf16";
+
+ INDEX_TO_CHARSET[56] = "utf16le";
+ INDEX_TO_CHARSET[62] = "utf16le";
+
+ INDEX_TO_CHARSET[57] = "cp1256";
+ INDEX_TO_CHARSET[67] = "cp1256";
+
+ INDEX_TO_CHARSET[29] = "cp1257";
+ INDEX_TO_CHARSET[58] = "cp1257";
+ INDEX_TO_CHARSET[59] = "cp1257";
+
+ INDEX_TO_CHARSET[60] = "utf32";
+ INDEX_TO_CHARSET[61] = "utf32";
+ for (int i = 160; i <= 183; i++) {
+ INDEX_TO_CHARSET[i] = "utf32";
+ }
+ INDEX_TO_CHARSET[391] = "utf32";
+
+ INDEX_TO_CHARSET[63] = "binary";
+
+ INDEX_TO_CHARSET[92] = "geostd8";
+ INDEX_TO_CHARSET[93] = "geostd8";
+
+ INDEX_TO_CHARSET[95] = "cp932";
+ INDEX_TO_CHARSET[96] = "cp932";
+
+ INDEX_TO_CHARSET[97] = "eucjpms";
+ INDEX_TO_CHARSET[98] = "eucjpms";
+
+ for (int i = 248; i <= 250; i++) {
+ INDEX_TO_CHARSET[i] = "gb18030";
+ }
+
+ // charset --> index
+ for (int i = 0; i < 2048; i++) {
+ String charset = INDEX_TO_CHARSET[i];
+ if (charset != null && CHARSET_TO_INDEX.get(charset) == null) {
+ CHARSET_TO_INDEX.put(charset, i);
+ }
+ }
+ CHARSET_TO_INDEX.put("iso-8859-1", 14);
+ CHARSET_TO_INDEX.put("iso_8859_1", 14);
+ CHARSET_TO_INDEX.put("utf-8", 33);
+ CHARSET_TO_INDEX.put("utf8mb4", 45);
+ }
+
+ public static final String getCharset(int index) {
+ return INDEX_TO_CHARSET[index];
+ }
+
+ public static final int getIndex(String charset) {
+ if (charset == null || charset.length() == 0) {
+ return 0;
+ } else {
+ Integer i = CHARSET_TO_INDEX.get(charset.toLowerCase());
+ return (i == null) ? 0 : i.intValue();
+ }
+ }
+
+ /**
+ * 'utf8' COLLATE 'utf8_general_ci'
+ *
+ * @param charset
+ * @return
+ */
+ public static final String collateCharset(String charset) {
+ String[] output = StringUtils.split(charset, "COLLATE");
+ return output[0].replace('\'', ' ').trim();
+ }
+
+ public static String getJavaCharset(String charset) {
+ if ("utf8".equals(charset)) {
+ return charset;
+ }
+
+ if (StringUtils.endsWithIgnoreCase(charset, "utf8mb4")) {
+ return "utf-8";
+ }
+
+ if (StringUtils.endsWithIgnoreCase(charset, "binary")) {
+ return "iso_8859_1";
+ }
+
+ return charset;
+ }
+}
diff --git a/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java
new file mode 100644
index 00000000..09837d09
--- /dev/null
+++ b/driver/src/test/java/com/alibaba/otter/canal/parse/driver/mysql/CharsetUtilTest.java
@@ -0,0 +1,51 @@
+package com.alibaba.otter.canal.parse.driver.mysql;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import com.alibaba.otter.canal.parse.driver.mysql.utils.CharsetUtil;
+
+public class CharsetUtilTest {
+
+ @Test
+ public void testLatin1() {
+ int charsetIndex = 5;
+ String charset = "latin1";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+
+ @Test
+ public void testGbk() {
+ int charsetIndex = 87;
+ String charset = "gbk";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+
+ @Test
+ public void testGb2312() {
+ int charsetIndex = 24;
+ String charset = "gb2312";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+
+ @Test
+ public void testUtf8() {
+ int charsetIndex = 213;
+ String charset = "utf8";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+
+ @Test
+ public void testUtf8mb4() {
+ int charsetIndex = 235;
+ String charset = "utf8mb4";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+
+ @Test
+ public void testBinary() {
+ int charsetIndex = 63;
+ String charset = "binary";
+ Assert.assertTrue(charset.equals(CharsetUtil.getCharset(charsetIndex)));
+ }
+}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
index dbb40287..8902c05e 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java
@@ -353,7 +353,11 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle
eventSink.interrupt();
if (multiStageCoprocessor != null && multiStageCoprocessor.isStart()) {
- multiStageCoprocessor.stop();
+ try {
+ multiStageCoprocessor.stop();
+ } catch (Throwable t) {
+ logger.debug("multi processor rejected:", t);
+ }
}
try {
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
index 8050186d..5e94e40b 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/AbstractMysqlEventParser.java
@@ -60,6 +60,7 @@ public abstract class AbstractMysqlEventParser extends AbstractEventParser {
convert.setFilterQueryDdl(filterQueryDdl);
convert.setFilterRows(filterRows);
convert.setFilterTableError(filterTableError);
+ convert.setUseDruidDdlFilter(useDruidDdlFilter);
return convert;
}
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 4348dc29..e3056cb9 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
@@ -37,21 +37,23 @@ import com.taobao.tddl.dbsync.binlog.LogBuffer;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
+import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
public class MysqlConnection implements ErosaConnection {
- private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
+ private static final Logger logger = LoggerFactory.getLogger(MysqlConnection.class);
private MysqlConnector connector;
private long slaveId;
- private Charset charset = Charset.forName("UTF-8");
+ private Charset charset = Charset.forName("UTF-8");
private BinlogFormat binlogFormat;
private BinlogImage binlogImage;
// tsdb releated
private AuthenticationInfo authInfo;
- protected int connTimeout = 5 * 1000; // 5秒
- protected int soTimeout = 60 * 60 * 1000; // 1小时
+ protected int connTimeout = 5 * 1000; // 5秒
+ protected int soTimeout = 60 * 60 * 1000; // 1小时
+ private int binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
// dump binlog bytes, 暂不包括meta与TSDB
private AtomicLong receivedBinlogBytes;
@@ -118,7 +120,7 @@ public class MysqlConnection implements ErosaConnection {
*/
public void seek(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
-
+ loadBinlogChecksum();
sendBinlogDump(binlogfilename, binlogPosition);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
fetcher.start(connector.getChannel());
@@ -128,6 +130,7 @@ public class MysqlConnection implements ErosaConnection {
decoder.handle(LogEvent.QUERY_EVENT);
decoder.handle(LogEvent.XID_EVENT);
LogContext context = new LogContext();
+ context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
accumulateReceivedBytes(fetcher.limit());
LogEvent event = null;
@@ -145,12 +148,14 @@ public class MysqlConnection implements ErosaConnection {
public void dump(String binlogfilename, Long binlogPosition, SinkFunction func) throws IOException {
updateSettings();
+ loadBinlogChecksum();
sendRegisterSlave();
sendBinlogDump(binlogfilename, binlogPosition);
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();
+ context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
while (fetcher.fetch()) {
accumulateReceivedBytes(fetcher.limit());
LogEvent event = null;
@@ -173,6 +178,7 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(GTIDSet gtidSet, SinkFunction func) throws IOException {
updateSettings();
+ loadBinlogChecksum();
sendBinlogDumpGTID(gtidSet);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
@@ -180,6 +186,7 @@ public class MysqlConnection implements ErosaConnection {
fetcher.start(connector.getChannel());
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
+ context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
// fix bug: #890 将gtid传输至context中,供decode使用
context.setGtidSet(gtidSet);
while (fetcher.fetch()) {
@@ -207,9 +214,11 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(String binlogfilename, Long binlogPosition, MultiStageCoprocessor coprocessor) throws IOException {
updateSettings();
+ loadBinlogChecksum();
sendRegisterSlave();
sendBinlogDump(binlogfilename, binlogPosition);
((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
+ ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
try {
fetcher.start(connector.getChannel());
@@ -234,9 +243,10 @@ public class MysqlConnection implements ErosaConnection {
@Override
public void dump(GTIDSet gtidSet, MultiStageCoprocessor coprocessor) throws IOException {
updateSettings();
+ loadBinlogChecksum();
sendBinlogDumpGTID(gtidSet);
-
((MysqlMultiStageCoprocessor) coprocessor).setConnection(this);
+ ((MysqlMultiStageCoprocessor) coprocessor).setBinlogChecksum(binlogChecksum);
DirectLogFetcher fetcher = new DirectLogFetcher(connector.getReceiveBufferSize());
try {
fetcher.start(connector.getChannel());
@@ -482,6 +492,52 @@ public class MysqlConnection implements ErosaConnection {
}
}
+ /**
+ * 获取主库checksum信息
+ *
+ *
+ * mariadb区别于mysql会在binlog的第一个事件Rotate_Event里也会采用checksum逻辑,而mysql是在第二个binlog事件之后才感知是否需要处理checksum
+ * 导致maraidb只要是开启checksum就会出现binlog文件名解析乱码
+ * fixed issue : https://github.com/alibaba/canal/issues/1081
+ *
+ */
+ private void loadBinlogChecksum() {
+ if (checkMariaDB()) {
+ ResultSetPacket rs = null;
+ try {
+ rs = query("select @@global.binlog_checksum");
+ } catch (IOException e) {
+ throw new CanalParseException(e);
+ }
+
+ List columnValues = rs.getFieldValues();
+ if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
+ binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
+ } else {
+ binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
+ }
+ }
+ }
+
+ /**
+ * 获取是否为mariadb
+ */
+ private boolean checkMariaDB() {
+ ResultSetPacket rs = null;
+ try {
+ rs = query("SELECT @@version");
+ } catch (IOException e) {
+ throw new CanalParseException(e);
+ }
+
+ List columnValues = rs.getFieldValues();
+ if (columnValues != null && columnValues.size() >= 1) {
+ return StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB");
+ }
+
+ return false;
+ }
+
private void accumulateReceivedBytes(long x) {
if (receivedBinlogBytes != null) {
receivedBinlogBytes.addAndGet(x);
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 df9c952c..e8ca45e3 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
@@ -69,6 +69,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
// update by yishun.chen,特殊异常处理参数
private int dumpErrorCount = 0; // binlogDump失败异常计数
private int dumpErrorCountThreshold = 2; // binlogDump失败异常计数阀值
+ private boolean rdsOssMode = false;
protected ErosaConnection buildErosaConnection() {
return buildMysqlConnection(this.runningInfo);
@@ -352,7 +353,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return logPosition.getPostion();
}
- if (masterPosition!=null && StringUtils.isNotEmpty(masterPosition.getGtid())) {
+ if (masterPosition != null && StringUtils.isNotEmpty(masterPosition.getGtid())) {
return masterPosition;
}
}
@@ -493,6 +494,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
dumpErrorCount = 0;
return findPosition;
}
+
+ Long timestamp = logPosition.getPostion().getTimestamp();
+ if (isRdsOssMode() && (timestamp != null && timestamp > 0)) {
+ // 如果binlog位点不存在,并且属于timestamp不为空,可以返回null走到oss binlog处理
+ return null;
+ }
}
// 其余情况
logger.warn("prepare to find start position just last position\n {}",
@@ -745,14 +752,13 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
logPosition.setPostion(entryPosition);
}
- if (entry == null) {
- return true;
- }
-
- String logfilename = entry.getHeader().getLogfileName();
- Long logfileoffset = entry.getHeader().getLogfileOffset();
- Long logposTimestamp = entry.getHeader().getExecuteTime();
- Long serverId = entry.getHeader().getServerId();
+ // 直接用event的位点来处理,解决一个binlog文件里没有任何事件导致死循环无法退出的问题
+ String logfilename = event.getHeader().getLogFileName();
+ // 记录的是binlog end offest,
+ // 因为与其对比的offest是show master status里的end offest
+ Long logfileoffset = event.getHeader().getLogPos();
+ Long logposTimestamp = event.getHeader().getWhen() * 1000;
+ Long serverId = event.getHeader().getServerId();
// 如果最小的一条记录都不满足条件,可直接退出
if (logposTimestamp >= startTimestamp) {
@@ -764,6 +770,10 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
return false;
}
+ if (entry == null) {
+ return true;
+ }
+
// 记录一下上一个事务结束的位置,即下一个事务的position
// position = current +
// data.length,代表该事务的下一条offest,避免多余的事务重复
@@ -905,4 +915,12 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
this.dumpErrorCountThreshold = dumpErrorCountThreshold;
}
+ public boolean isRdsOssMode() {
+ return rdsOssMode;
+ }
+
+ public void setRdsOssMode(boolean rdsOssMode) {
+ this.rdsOssMode = rdsOssMode;
+ }
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
index 3db8ecc4..703de818 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlMultiStageCoprocessor.java
@@ -33,6 +33,7 @@ import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
@@ -69,6 +70,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private WorkerPool workerPool;
private BatchEventProcessor simpleParserStage;
private BatchEventProcessor sinkStoreStage;
+ private LogContext logContext;
public MysqlMultiStageCoprocessor(int ringBufferSize, int parserThreadCount, LogEventConvert logEventConvert,
EventTransactionBuffer transactionBuffer, String destination){
@@ -95,9 +97,10 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
SequenceBarrier sequenceBarrier = disruptorMsgBuffer.newBarrier();
ExceptionHandler exceptionHandler = new SimpleFatalExceptionHandler();
// stage 2
+ this.logContext = new LogContext();
simpleParserStage = new BatchEventProcessor(disruptorMsgBuffer,
sequenceBarrier,
- new SimpleParserStage());
+ new SimpleParserStage(logContext));
simpleParserStage.setExceptionHandler(exceptionHandler);
disruptorMsgBuffer.addGatingSequences(simpleParserStage.getSequence());
@@ -128,6 +131,12 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
workerPool.start(parserExecutor);
}
+ public void setBinlogChecksum(int binlogChecksum) {
+ if (binlogChecksum != LogEvent.BINLOG_CHECKSUM_ALG_OFF) {
+ logContext.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
+ }
+ }
+
@Override
public void stop() {
// fix bug #968,对于pool与
@@ -239,9 +248,9 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
private LogDecoder decoder;
private LogContext context;
- public SimpleParserStage(){
+ public SimpleParserStage(LogContext context){
decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
- context = new LogContext();
+ this.context = context;
if (gtidSet != null) {
context.setGtidSet(gtidSet);
}
@@ -266,6 +275,7 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
needDmlParse = true;
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
tableMeta = logEventConvert.parseRowsEventForTableMeta((UpdateRowsLogEvent) logEvent);
needDmlParse = true;
@@ -468,4 +478,5 @@ public class MysqlMultiStageCoprocessor extends AbstractCanalLifeCycle implement
public void setGtidSet(GTIDSet gtidSet) {
this.gtidSet = gtidSet;
}
+
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
index ec71465b..5c5432c6 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java
@@ -120,6 +120,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
case LogEvent.WRITE_ROWS_EVENT:
return parseRowsEvent((WriteRowsLogEvent) logEvent);
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
return parseRowsEvent((UpdateRowsLogEvent) logEvent);
case LogEvent.DELETE_ROWS_EVENT_V1:
@@ -260,7 +261,9 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
return null;
}
- if (!isSeek) {
+ boolean isDml = (type == EventType.INSERT || type == EventType.UPDATE || type == EventType.DELETE);
+
+ if (!isSeek && !isDml) {
// 使用新的表结构元数据管理方式
EntryPosition position = createPosition(event.getHeader());
tableMetaCache.apply(position, event.getDbName(), queryString, null);
@@ -268,8 +271,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
Header header = createHeader(event.getHeader(), schemaName, tableName, type);
RowChange.Builder rowChangeBuider = RowChange.newBuilder();
- if (type != EventType.QUERY && type != EventType.INSERT && type != EventType.UPDATE
- && type != EventType.DELETE) {
+ if (type != EventType.QUERY && !isDml) {
rowChangeBuider.setIsDdl(true);
}
rowChangeBuider.setSql(queryString);
@@ -503,7 +505,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
int type = event.getHeader().getType();
if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
eventType = EventType.INSERT;
- } else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type) {
+ } else if (LogEvent.UPDATE_ROWS_EVENT_V1 == type || LogEvent.UPDATE_ROWS_EVENT == type
+ || LogEvent.PARTIAL_UPDATE_ROWS_EVENT == type) {
eventType = EventType.UPDATE;
} else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
eventType = EventType.DELETE;
@@ -522,7 +525,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
boolean tableError = false;
int rowsCount = 0;
- while (buffer.nextOneRow(columns)) {
+ while (buffer.nextOneRow(columns, false)) {
// 处理row记录
RowData.Builder rowDataBuilder = RowData.newBuilder();
if (EventType.INSERT == eventType) {
@@ -534,7 +537,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
} else {
// update需要处理before/after
tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
- if (!buffer.nextOneRow(changeColumns)) {
+ if (!buffer.nextOneRow(changeColumns, true)) {
rowChangeBuider.addRowDatas(rowDataBuilder.build());
break;
}
@@ -567,8 +570,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
}
private EntryPosition createPosition(LogHeader logHeader) {
- return new EntryPosition(logHeader.getLogFileName(),
- logHeader.getLogPos(),
+ return new EntryPosition(logHeader.getLogFileName(), logHeader.getLogPos() - logHeader.getEventLen(), // startPos
logHeader.getWhen() * 1000L,
logHeader.getServerId()); // 记录到秒
}
@@ -577,7 +579,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
boolean isAfter, TableMeta tableMeta) throws UnsupportedEncodingException {
int columnCnt = event.getTable().getColumnCnt();
ColumnInfo[] columnInfo = event.getTable().getColumnInfo();
-
+ // mysql8.0针对set @@global.binlog_row_metadata='FULL' 可以记录部分的metadata信息
+ boolean existOptionalMetaData = event.getTable().isExistOptionalMetaData();
boolean tableError = false;
// check table fileds count,只能处理加字段
boolean existRDSNoPrimaryKey = false;
@@ -631,22 +634,43 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
continue;
}
- if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
- // 不解析最后一列
- buffer.nextValue(info.type, info.meta, false);
- continue;
- }
-
- Column.Builder columnBuilder = Column.newBuilder();
-
FieldMeta fieldMeta = null;
if (tableMeta != null && !tableError) {
// 处理file meta
fieldMeta = tableMeta.getFields().get(i);
+ }
+
+ if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
+ // 不解析最后一列
+ buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, false);
+ continue;
+ }
+
+ if (fieldMeta != null && existOptionalMetaData) {
+ // check column info
+ boolean check = StringUtils.equalsIgnoreCase(fieldMeta.getColumnName(), info.name);
+ check &= (fieldMeta.isUnsigned() == info.unsigned);
+ check &= (fieldMeta.isNullable() == info.nullable);
+
+ if (!check) {
+ throw new CanalParseException("MySQL8.0 unmatch column metadata & pls submit issue , db : "
+ + fieldMeta.toString() + " , binlog : " + info.toString()
+ + " , on : " + event.getHeader().getLogFileName() + ":"
+ + event.getHeader().getLogPos());
+ }
+ }
+
+ Column.Builder columnBuilder = Column.newBuilder();
+ if (fieldMeta != null) {
columnBuilder.setName(fieldMeta.getColumnName());
columnBuilder.setIsKey(fieldMeta.isKey());
// 增加mysql type类型,issue 73
columnBuilder.setMysqlType(fieldMeta.getColumnType());
+ } else if (existOptionalMetaData) {
+ columnBuilder.setName(info.name);
+ columnBuilder.setIsKey(info.pk);
+ // mysql8.0里没有mysql type类型
+ // columnBuilder.setMysqlType(fieldMeta.getColumnType());
}
columnBuilder.setIndex(i);
columnBuilder.setIsNull(false);
@@ -664,12 +688,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
isSingleBit = true;
}
}
- buffer.nextValue(info.type, info.meta, isBinary);
- if (existRDSNoPrimaryKey && i == columnCnt - 1 && info.type == LogEvent.MYSQL_TYPE_LONGLONG) {
- // 不解析最后一列
- continue;
- }
+ buffer.nextValue(fieldMeta.getColumnName(), i, info.type, info.meta, isBinary);
int javaType = buffer.getJavaType();
if (buffer.isNull()) {
columnBuilder.setIsNull(true);
@@ -683,7 +703,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
case Types.BIGINT:
// 处理unsigned类型
Number number = (Number) value;
- if (fieldMeta != null && fieldMeta.isUnsigned() && number.longValue() < 0) {
+ boolean isUnsigned = (fieldMeta != null ? fieldMeta.isUnsigned() : (existOptionalMetaData ? info.unsigned : false));
+ if (isUnsigned && number.longValue() < 0) {
switch (buffer.getLength()) {
case 1: /* MYSQL_TYPE_TINY */
columnBuilder.setValue(String.valueOf(Integer.valueOf(TINYINT_MAX_VALUE
@@ -772,7 +793,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
default:
columnBuilder.setValue(value.toString());
}
-
}
columnBuilder.setSqlType(javaType);
@@ -818,6 +838,7 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
Header.Builder headerBuilder = Header.newBuilder();
headerBuilder.setVersion(version);
headerBuilder.setLogfileName(logHeader.getLogFileName());
+ // 记录的是该binlog的start offest
headerBuilder.setLogfileOffset(logHeader.getLogPos() - logHeader.getEventLen());
headerBuilder.setServerId(logHeader.getServerId());
headerBuilder.setServerenCode(UTF_8);// 经过java输出后所有的编码为unicode
@@ -981,4 +1002,8 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
public void setFilterRows(boolean filterRows) {
this.filterRows = filterRows;
}
+
+ public void setUseDruidDdlFilter(boolean useDruidDdlFilter) {
+ this.useDruidDdlFilter = useDruidDdlFilter;
+ }
}
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
index e132514b..b90b89d5 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/BinlogDownloadQueue.java
@@ -41,6 +41,7 @@ import org.apache.http.ssl.TrustStrategy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.mysql.rds.data.BinlogFile;
/**
@@ -103,6 +104,9 @@ public class BinlogDownloadQueue {
public BinlogFile tryOne() throws Throwable {
BinlogFile binlogFile = binlogList.poll();
+ if (binlogFile == null) {
+ throw new CanalParseException("download binlog is null");
+ }
download(binlogFile);
hostId = binlogFile.getHostInstanceID();
this.currentSize++;
@@ -131,7 +135,7 @@ public class BinlogDownloadQueue {
if (StringUtils.isNotEmpty(needCompareName) && StringUtils.endsWith(needCompareName, "tar")) {
needCompareName = needCompareName.substring(0, needCompareName.indexOf("."));
}
- return fileName.equalsIgnoreCase(needCompareName) && binlogList.isEmpty();
+ return (needCompareName == null || fileName.equalsIgnoreCase(needCompareName)) && binlogList.isEmpty();
}
public void prepare() throws InterruptedException {
@@ -162,6 +166,14 @@ public class BinlogDownloadQueue {
this.currentSize = 0;
binlogList.clear();
downloadQueue.clear();
+ try {
+ downloadThread.interrupt();
+ downloadThread.join();// 等待其结束
+ } catch (InterruptedException e) {
+ // ignore
+ } finally {
+ downloadThread = null;
+ }
}
private void download(BinlogFile binlogFile) throws Throwable {
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
index 3c38ed0c..f3a220b0 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsBinlogEventParserProxy.java
@@ -22,14 +22,14 @@ import com.alibaba.otter.canal.parse.inbound.mysql.MysqlEventParser;
*/
public class RdsBinlogEventParserProxy extends MysqlEventParser {
- private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
- private String accesskey; // 云账号的ak
- private String secretkey; // 云账号sk
- private String instanceId; // rds实例id
- private String directory; // binlog目录
- private int batchFileSize = 4; // 最多下载的binlog文件数量
+ private String rdsOpenApiUrl = "https://rds.aliyuncs.com/"; // openapi地址
+ private String accesskey; // 云账号的ak
+ private String secretkey; // 云账号sk
+ private String instanceId; // rds实例id
+ private String directory; // binlog目录
+ private int batchFileSize = 4; // 最多下载的binlog文件数量
- private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser();
+ private RdsLocalBinlogEventParser rdsLocalBinlogEventParser = null;
private ExecutorService executorService = Executors.newSingleThreadExecutor(new ThreadFactory() {
@Override
@@ -43,8 +43,11 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
@Override
public void start() {
- if (StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey)
+ if (rdsLocalBinlogEventParser == null && StringUtils.isNotEmpty(accesskey) && StringUtils.isNotEmpty(secretkey)
&& StringUtils.isNotEmpty(instanceId)) {
+ rdsLocalBinlogEventParser = new RdsLocalBinlogEventParser();
+ // rds oss mode
+ setRdsOssMode(true);
final ParserExceptionHandler targetHandler = this.getParserExceptionHandler();
if (directory == null) {
directory = System.getProperty("java.io.tmpdir", "/tmp") + "/" + destination;
@@ -119,10 +122,18 @@ public class RdsBinlogEventParserProxy extends MysqlEventParser {
long serverId = rdsBinlogEventParserProxy.getServerId();
rdsLocalBinlogEventParser.setServerId(serverId);
rdsBinlogEventParserProxy.stop();
+ } catch (Throwable e) {
+ logger.info("handle exception failed", e);
+ }
+
+ try {
logger.info("start rds mysql binlog parser!");
rdsLocalBinlogEventParser.start();
} catch (Throwable e) {
logger.info("handle exception failed", e);
+ rdsLocalBinlogEventParser.stop();
+ RdsBinlogEventParserProxy rdsBinlogEventParserProxy = RdsBinlogEventParserProxy.this;
+ rdsBinlogEventParserProxy.start();// 继续重试
}
}
});
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
index aac14d99..2e0960fe 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/rds/RdsLocalBinlogEventParser.java
@@ -57,7 +57,11 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
if (entryPosition == null) {
throw new PositionNotFoundException("position not found!");
}
- long startTimeInMill = entryPosition.getTimestamp();
+ Long startTimeInMill = entryPosition.getTimestamp();
+ if (startTimeInMill == null || startTimeInMill <= 0) {
+ throw new PositionNotFoundException("position timestamp is empty!");
+ }
+
startTime = startTimeInMill;
List binlogFiles = RdsBinlogOpenApi.listBinlogFiles(url,
accesskey,
@@ -65,6 +69,10 @@ public class RdsLocalBinlogEventParser extends LocalBinlogEventParser implements
instanceId,
new Date(startTime),
new Date(endTime));
+ if (binlogFiles.isEmpty()) {
+ throw new CanalParseException("start timestamp : " + startTimeInMill + " binlog files is empty");
+ }
+
binlogDownloadQueue = new BinlogDownloadQueue(binlogFiles, batchFileSize, directory);
binlogDownloadQueue.silenceDownload();
needWait = true;
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
index b0dd89ce..8896e2f5 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/DatabaseTableMeta.java
@@ -7,6 +7,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
@@ -48,18 +49,26 @@ public class DatabaseTableMeta implements TableMetaTSDB {
private static Logger logger = LoggerFactory.getLogger(DatabaseTableMeta.class);
private static Pattern pattern = Pattern.compile("Duplicate entry '.*' for key '*'");
private static Pattern h2Pattern = Pattern.compile("Unique index or primary key violation");
+ private static ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
+ @Override
+ public Thread newThread(Runnable r) {
+ Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]");
+ thread.setDaemon(true);
+ return thread;
+ }
+ });
private String destination;
private MemoryTableMeta memoryTableMeta;
private MysqlConnection connection; // 查询meta信息的链接
private CanalEventFilter filter;
private CanalEventFilter blackFilter;
private EntryPosition lastPosition;
- private ScheduledExecutorService scheduler;
private MetaHistoryDAO metaHistoryDAO;
private MetaSnapshotDAO metaSnapshotDAO;
private int snapshotInterval = 24;
private int snapshotExpire = 360;
-
+ private ScheduledFuture> scheduleSnapshotFuture;
+
public DatabaseTableMeta(){
}
@@ -68,19 +77,10 @@ public class DatabaseTableMeta implements TableMetaTSDB {
public boolean init(final String destination) {
this.destination = destination;
this.memoryTableMeta = new MemoryTableMeta();
- this.scheduler = Executors.newSingleThreadScheduledExecutor(new ThreadFactory() {
-
- @Override
- public Thread newThread(Runnable r) {
- Thread thread = new Thread(r, "[scheduler-table-meta-snapshot]");
- thread.setDaemon(true);
- return thread;
- }
- });
// 24小时生成一份snapshot
if (snapshotInterval > 0) {
- scheduler.scheduleWithFixedDelay(new Runnable() {
+ scheduleSnapshotFuture = scheduler.scheduleWithFixedDelay(new Runnable() {
@Override
public void run() {
@@ -105,6 +105,26 @@ public class DatabaseTableMeta implements TableMetaTSDB {
}
return true;
}
+
+ @Override
+ public void destory() {
+ if (memoryTableMeta != null) {
+ memoryTableMeta.destory();
+ }
+
+ if (connection != null) {
+ try {
+ connection.disconnect();
+ } catch (IOException e) {
+ logger.error("ERROR # disconnect meta connection for address:{}", connection.getConnector()
+ .getAddress(), e);
+ }
+ }
+
+ if (scheduleSnapshotFuture != null) {
+ scheduleSnapshotFuture.cancel(false);
+ }
+ }
@Override
public TableMeta find(String schema, String table) {
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
index 904b55a9..c53e1a08 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java
@@ -58,6 +58,11 @@ public class MemoryTableMeta implements TableMetaTSDB {
public boolean init(String destination) {
return true;
}
+
+ @Override
+ public void destory() {
+ tableMetas.clear();
+ }
public boolean apply(EntryPosition position, String schema, String ddl, String extra) {
tableMetas.clear();
diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java
index d712adbe..f968ae12 100644
--- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java
+++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/TableMetaTSDB.java
@@ -18,6 +18,11 @@ public interface TableMetaTSDB {
*/
public boolean init(String destination);
+ /**
+ * 销毁资源
+ */
+ public void destory();
+
/**
* 获取当前的表结构
*/
diff --git a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
index aa48f9a3..fd01e6a8 100644
--- a/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
+++ b/parse/src/test/java/com/alibaba/otter/canal/parse/DirectLogFetcherTest.java
@@ -1,7 +1,12 @@
package com.alibaba.otter.canal.parse;
import java.io.IOException;
+import java.io.Serializable;
+import java.io.UnsupportedEncodingException;
import java.net.InetSocketAddress;
+import java.nio.charset.Charset;
+import java.util.BitSet;
+import java.util.List;
import org.apache.commons.lang.StringUtils;
import org.junit.Assert;
@@ -10,29 +15,49 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlConnector;
+import com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.MysqlUpdateExecutor;
import com.alibaba.otter.canal.parse.driver.mysql.packets.HeaderPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.BinlogDumpCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.client.RegisterSlaveCommandPacket;
import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ErrorPacket;
+import com.alibaba.otter.canal.parse.driver.mysql.packets.server.ResultSetPacket;
import com.alibaba.otter.canal.parse.driver.mysql.utils.PacketManager;
+import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.inbound.mysql.dbsync.DirectLogFetcher;
import com.taobao.tddl.dbsync.binlog.LogContext;
import com.taobao.tddl.dbsync.binlog.LogDecoder;
import com.taobao.tddl.dbsync.binlog.LogEvent;
+import com.taobao.tddl.dbsync.binlog.event.DeleteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.FormatDescriptionLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.QueryLogEvent;
import com.taobao.tddl.dbsync.binlog.event.RotateLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.RowsLogBuffer;
+import com.taobao.tddl.dbsync.binlog.event.RowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.RowsQueryLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.TableMapLogEvent.ColumnInfo;
+import com.taobao.tddl.dbsync.binlog.event.UpdateRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.WriteRowsLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.XidLogEvent;
+import com.taobao.tddl.dbsync.binlog.event.mariadb.AnnotateRowsEvent;
public class DirectLogFetcherTest {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+ protected String binlogFileName = "mysql-bin.000001";
+ protected Charset charset = Charset.forName("utf-8");
+ private boolean isMariaDB;
+ private int binlogChecksum;
@Test
public void testSimple() {
DirectLogFetcher fetcher = new DirectLogFetcher();
try {
- MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "xxxx", "xxxx");
+ MysqlConnector connector = new MysqlConnector(new InetSocketAddress("127.0.0.1", 3306), "root", "hello");
connector.connect();
updateSettings(connector);
+ loadBinlogChecksum(connector);
sendRegisterSlave(connector, 3);
sendBinlogDump(connector, "mysql-bin.000001", 4L, 3);
@@ -40,6 +65,8 @@ public class DirectLogFetcherTest {
LogDecoder decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
LogContext context = new LogContext();
+ context.setFormatDescription(new FormatDescriptionLogEvent(4, binlogChecksum));
+
while (fetcher.fetch()) {
LogEvent event = null;
event = decoder.decode(fetcher, context);
@@ -57,21 +84,22 @@ public class DirectLogFetcherTest {
break;
case LogEvent.WRITE_ROWS_EVENT_V1:
case LogEvent.WRITE_ROWS_EVENT:
- // parseRowsEvent((WriteRowsLogEvent) event);
+ parseRowsEvent((WriteRowsLogEvent) event);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
- // parseRowsEvent((UpdateRowsLogEvent) event);
+ parseRowsEvent((UpdateRowsLogEvent) event);
break;
case LogEvent.DELETE_ROWS_EVENT_V1:
case LogEvent.DELETE_ROWS_EVENT:
- // parseRowsEvent((DeleteRowsLogEvent) event);
+ parseRowsEvent((DeleteRowsLogEvent) event);
break;
case LogEvent.QUERY_EVENT:
- // parseQueryEvent((QueryLogEvent) event);
+ parseQueryEvent((QueryLogEvent) event);
break;
case LogEvent.ROWS_QUERY_LOG_EVENT:
- // parseRowsQueryEvent((RowsQueryLogEvent) event);
+ parseRowsQueryEvent((RowsQueryLogEvent) event);
break;
case LogEvent.ANNOTATE_ROWS_EVENT:
break;
@@ -188,9 +216,141 @@ public class DirectLogFetcherTest {
}
}
+ private void loadBinlogChecksum(MysqlConnector connector) {
+ checkMariaDB(connector);
+ if (isMariaDB) {
+ ResultSetPacket rs = null;
+ try {
+ rs = query("select @@global.binlog_checksum", connector);
+ } catch (IOException e) {
+ throw new CanalParseException(e);
+ }
+
+ List columnValues = rs.getFieldValues();
+ if (columnValues != null && columnValues.size() >= 1 && columnValues.get(0).toUpperCase().equals("CRC32")) {
+ binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_CRC32;
+ } else {
+ binlogChecksum = LogEvent.BINLOG_CHECKSUM_ALG_OFF;
+ }
+ }
+ }
+
+ private void checkMariaDB(MysqlConnector connector) {
+ ResultSetPacket rs = null;
+ try {
+ rs = query("SELECT @@version", connector);
+ } catch (IOException e) {
+ throw new CanalParseException(e);
+ }
+
+ List columnValues = rs.getFieldValues();
+ if (columnValues != null && columnValues.size() >= 1) {
+ isMariaDB = StringUtils.containsIgnoreCase(columnValues.get(0), "MariaDB");
+ }
+ }
+
+ public ResultSetPacket query(String cmd, MysqlConnector connector) throws IOException {
+ MysqlQueryExecutor exector = new MysqlQueryExecutor(connector);
+ return exector.query(cmd);
+ }
+
public void update(String cmd, MysqlConnector connector) throws IOException {
MysqlUpdateExecutor exector = new MysqlUpdateExecutor(connector);
exector.update(cmd);
}
+ protected void parseQueryEvent(QueryLogEvent event) {
+ System.out.println(String.format("================> binlog[%s:%s] , name[%s]",
+ binlogFileName,
+ event.getHeader().getLogPos() - event.getHeader().getEventLen(),
+ event.getCatalog()));
+ System.out.println("sql : " + event.getQuery());
+ }
+
+ protected void parseRowsQueryEvent(RowsQueryLogEvent event) throws Exception {
+ System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
+ .getLogPos() - event.getHeader().getEventLen()));
+ System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name()));
+ }
+
+ protected void parseAnnotateRowsEvent(AnnotateRowsEvent event) throws Exception {
+ System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
+ .getLogPos() - event.getHeader().getEventLen()));
+ System.out.println("sql : " + new String(event.getRowsQuery().getBytes("ISO-8859-1"), charset.name()));
+ }
+
+ protected void parseXidEvent(XidLogEvent event) throws Exception {
+ System.out.println(String.format("================> binlog[%s:%s]", binlogFileName, event.getHeader()
+ .getLogPos() - event.getHeader().getEventLen()));
+ System.out.println("xid : " + event.getXid());
+ }
+
+ protected void parseRowsEvent(RowsLogEvent event) {
+ try {
+ System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s]",
+ binlogFileName,
+ event.getHeader().getLogPos() - event.getHeader().getEventLen(),
+ event.getTable().getDbName(),
+ event.getTable().getTableName()));
+ RowsLogBuffer buffer = event.getRowsBuf(charset.name());
+ BitSet columns = event.getColumns();
+ BitSet changeColumns = event.getChangeColumns();
+ while (buffer.nextOneRow(columns)) {
+ // 处理row记录
+ int type = event.getHeader().getType();
+ if (LogEvent.WRITE_ROWS_EVENT_V1 == type || LogEvent.WRITE_ROWS_EVENT == type) {
+ // insert的记录放在before字段中
+ parseOneRow(event, buffer, columns, true);
+ } else if (LogEvent.DELETE_ROWS_EVENT_V1 == type || LogEvent.DELETE_ROWS_EVENT == type) {
+ // delete的记录放在before字段中
+ parseOneRow(event, buffer, columns, false);
+ } else {
+ // update需要处理before/after
+ System.out.println("-------> before");
+ parseOneRow(event, buffer, columns, false);
+ if (!buffer.nextOneRow(changeColumns, true)) {
+ break;
+ }
+ System.out.println("-------> after");
+ parseOneRow(event, buffer, changeColumns, true);
+ }
+
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("parse row data failed.", e);
+ }
+ }
+
+ protected void parseOneRow(RowsLogEvent event, RowsLogBuffer buffer, BitSet cols, boolean isAfter)
+ throws UnsupportedEncodingException {
+ TableMapLogEvent map = event.getTable();
+ if (map == null) {
+ throw new RuntimeException("not found TableMap with tid=" + event.getTableId());
+ }
+
+ final int columnCnt = map.getColumnCnt();
+ final ColumnInfo[] columnInfo = map.getColumnInfo();
+
+ for (int i = 0; i < columnCnt; i++) {
+ if (!cols.get(i)) {
+ continue;
+ }
+
+ ColumnInfo info = columnInfo[i];
+ buffer.nextValue(null, i, info.type, info.meta);
+
+ if (buffer.isNull()) {
+ //
+ } else {
+ final Serializable value = buffer.getValue();
+ if (value instanceof byte[]) {
+ System.out.println(new String((byte[]) value));
+ } else {
+ System.out.println(value);
+ }
+ }
+ }
+
+ }
+
}
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 79b88375..70b2dbd7 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
@@ -89,6 +89,7 @@ public class MysqlBinlogParsePerformanceTest {
parseRowsEvent((WriteRowsLogEvent) event, sum);
break;
case LogEvent.UPDATE_ROWS_EVENT_V1:
+ case LogEvent.PARTIAL_UPDATE_ROWS_EVENT:
case LogEvent.UPDATE_ROWS_EVENT:
parseRowsEvent((UpdateRowsLogEvent) event, sum);
break;
@@ -154,7 +155,7 @@ public class MysqlBinlogParsePerformanceTest {
parseOneRow(event, buffer, columns, false);
} else {
parseOneRow(event, buffer, columns, false);
- if (!buffer.nextOneRow(changeColumns)) {
+ if (!buffer.nextOneRow(changeColumns, true)) {
break;
}
parseOneRow(event, buffer, changeColumns, true);
@@ -182,7 +183,7 @@ public class MysqlBinlogParsePerformanceTest {
}
ColumnInfo info = columnInfo[i];
- buffer.nextValue(info.type, info.meta);
+ buffer.nextValue(null, i, info.type, info.meta);
if (buffer.isNull()) {
} else {
buffer.getValue();
diff --git a/parse/src/test/resources/tsdb/h2-tsdb.xml b/parse/src/test/resources/tsdb/h2-tsdb.xml
index c0a3d385..dcb6bc47 100644
--- a/parse/src/test/resources/tsdb/h2-tsdb.xml
+++ b/parse/src/test/resources/tsdb/h2-tsdb.xml
@@ -7,7 +7,7 @@
default-autowire="byName">
-
+
diff --git a/parse/src/test/resources/tsdb/mysql-tsdb.xml b/parse/src/test/resources/tsdb/mysql-tsdb.xml
index 099c034d..44f97203 100644
--- a/parse/src/test/resources/tsdb/mysql-tsdb.xml
+++ b/parse/src/test/resources/tsdb/mysql-tsdb.xml
@@ -7,7 +7,7 @@
default-autowire="byName">
-
+
diff --git a/pom.xml b/pom.xml
index 5cb51eac..0bb4bc11 100644
--- a/pom.xml
+++ b/pom.xml
@@ -371,6 +371,7 @@
+
+ -->
src/main/java
src/test/java
diff --git a/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java
index 665dbf55..fda5dc65 100644
--- a/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/common/CanalMessageSerializer.java
@@ -16,7 +16,7 @@ import com.google.protobuf.WireFormat;
public class CanalMessageSerializer {
@SuppressWarnings("deprecation")
- public static byte[] serializer(Message data) {
+ public static byte[] serializer(Message data, boolean filterTransactionEntry) {
try {
if (data != null) {
if (data.getId() != -1) {
@@ -53,8 +53,14 @@ public class CanalMessageSerializer {
output.checkNoSpaceLeft();
return body;
} else if (!CollectionUtils.isEmpty(data.getEntries())) {
+ // mq模式只会走到非rowEntry模式
CanalPacket.Messages.Builder messageBuilder = CanalPacket.Messages.newBuilder();
for (CanalEntry.Entry entry : data.getEntries()) {
+ if (filterTransactionEntry
+ && (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND)) {
+ continue;
+ }
+
messageBuilder.addMessages(entry.toByteString());
}
diff --git a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java
index 77caa0a6..39c297d8 100644
--- a/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/kafka/MessageSerializer.java
@@ -2,6 +2,7 @@ package com.alibaba.otter.canal.kafka;
import java.util.Map;
+import org.apache.commons.lang.BooleanUtils;
import org.apache.kafka.common.serialization.Serializer;
import com.alibaba.otter.canal.common.CanalMessageSerializer;
@@ -15,13 +16,20 @@ import com.alibaba.otter.canal.protocol.Message;
*/
public class MessageSerializer implements Serializer {
+ private boolean filterTransactionEntry = false;
+
+ public MessageSerializer(){
+ this.filterTransactionEntry = BooleanUtils.toBoolean(System.getProperty("canal.instance.filter.transaction.entry",
+ "false"));
+ }
+
@Override
public void configure(Map configs, boolean isKey) {
}
@Override
public byte[] serialize(String topic, Message data) {
- return CanalMessageSerializer.serializer(data);
+ return CanalMessageSerializer.serializer(data, filterTransactionEntry);
}
@Override
diff --git a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
index aaace132..6173cebd 100644
--- a/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
+++ b/server/src/main/java/com/alibaba/otter/canal/rocketmq/CanalRocketMQProducer.java
@@ -47,7 +47,8 @@ public class CanalRocketMQProducer implements CanalMQProducer {
Callback callback) {
if (!mqProperties.getFlatMessage()) {
try {
- Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data));
+ Message message = new Message(destination.getTopic(), CanalMessageSerializer.serializer(data,
+ mqProperties.isFilterTransactionEntry()));
logger.debug("send message:{} to destination:{}, partition: {}",
message,
destination.getCanalDestination(),
@@ -77,8 +78,8 @@ public class CanalRocketMQProducer implements CanalMQProducer {
JSON.toJSONString(flatMessage),
destination.getTopic(),
destination.getPartition());
- Message message = new Message(destination.getTopic(), JSON.toJSONString(flatMessage)
- .getBytes());
+ Message message = new Message(destination.getTopic(),
+ JSON.toJSONString(flatMessage).getBytes());
this.defaultMQProducer.send(message, new MessageQueueSelector() {
@Override
@@ -98,28 +99,32 @@ public class CanalRocketMQProducer implements CanalMQProducer {
int length = partitionFlatMessage.length;
for (int i = 0; i < length; i++) {
FlatMessage flatMessagePart = partitionFlatMessage[i];
- logger.debug("flatMessagePart: {}, partition: {}",
- JSON.toJSONString(flatMessagePart),
- i);
- final int index = i;
- try {
- Message message = new Message(destination.getTopic(),
- JSON.toJSONString(flatMessagePart).getBytes());
- this.defaultMQProducer.send(message, new MessageQueueSelector() {
+ if (flatMessagePart != null) {
+ logger.debug("flatMessagePart: {}, partition: {}",
+ JSON.toJSONString(flatMessagePart),
+ i);
+ final int index = i;
+ try {
+ Message message = new Message(destination.getTopic(),
+ JSON.toJSONString(flatMessagePart).getBytes());
+ this.defaultMQProducer.send(message, new MessageQueueSelector() {
- @Override
- public MessageQueue select(List mqs, Message msg, Object arg) {
- if (index > mqs.size()) {
- throw new CanalServerException("partition number is error,config num:"
- + destination.getPartitionsNum()
- + ", mq num: " + mqs.size());
+ @Override
+ public MessageQueue select(List mqs, Message msg,
+ Object arg) {
+ if (index > mqs.size()) {
+ throw new CanalServerException(
+ "partition number is error,config num:"
+ + destination.getPartitionsNum()
+ + ", mq num: " + mqs.size());
+ }
+ return mqs.get(index);
}
- return mqs.get(index);
- }
- }, null);
- } catch (Exception e) {
- logger.error("send flat message to hashed partition error", e);
- callback.rollback();
+ }, null);
+ } catch (Exception e) {
+ logger.error("send flat message to hashed partition error", e);
+ callback.rollback();
+ }
}
}
}