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

merge and support polardb-x recycle_bin binlog format

Browse Source
This commit is contained in:
jianghang.loujh
2025-12-25 11:47:36 +08:00
parent 5a5e60d371
commit d18bde71c6
2 changed files with 108 additions and 140 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/prometheus/PrometheusService.java
View File

@@ -1,14 +1,15 @@
package com.alibaba.otter.canal.adapter.launcher.prometheus; package com.alibaba.otter.canal.adapter.launcher.prometheus;
import io.prometheus.client.exporter.HTTPServer; import java.io.IOException;
import io.prometheus.client.hotspot.DefaultExports;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;
import java.io.IOException; import io.prometheus.client.exporter.HTTPServer;
import io.prometheus.client.hotspot.DefaultExports;
/** /**
* @author sunxien * @author sunxien
@@ -25,8 +26,9 @@ public class PrometheusService implements InitializingBean, DisposableBean {
/** /**
* <pre> * <pre>
* Canal Server admin port: 11112 * Canal Server admin port: 11110
* Canal Server tcp port: 11111 * Canal Server tcp port: 11111
* Canal Server metric port: 11112
* Canal Adapter metric port: 11113 * Canal Adapter metric port: 11113
* </pre> * </pre>
*/ */

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

@@ -3,6 +3,7 @@ package com.taobao.tddl.dbsync.binlog.event;
import java.io.IOException; import java.io.IOException;
import java.math.BigInteger; import java.math.BigInteger;
import java.nio.charset.Charset; import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import com.taobao.tddl.dbsync.binlog.CharsetConversion; import com.taobao.tddl.dbsync.binlog.CharsetConversion;
import com.taobao.tddl.dbsync.binlog.LogBuffer; import com.taobao.tddl.dbsync.binlog.LogBuffer;
@@ -281,82 +282,28 @@ public class QueryLogEvent extends LogEvent {
* packet (i.e. a query) sent from client to master; First, an auxiliary * packet (i.e. a query) sent from client to master; First, an auxiliary
* log_event status vars estimation: * log_event status vars estimation:
*/ */
public static final int MAX_SIZE_LOG_EVENT_STATUS = (1 + 4 /* type, flags2 */ public static final int MAX_SIZE_LOG_EVENT_STATUS = (1 + 4 /* type, flags2 */ + 1 + 8 /* type,sql_mode */ + 1 + 1 + 255 /* type,length, catalog */
+ 1 + 8 /* + 1 + 4 /* type,auto_increment */
* type, + 1 + 6 /* type, charset */
* sql_mode + 1 + 1 + 255 /* type, length, time_zone */
*/ + 1 + 2 /* type, lc_time_names_number */
+ 1 + 1 + 255/* + 1 + 2 /* type, charset_database_number */
* type, + 1 + 8 /* type, table_map_for_update */
* length + 1 + 4 /* type,master_data_written */
* , + 1 + 1 + 32 * 3 /* type, user_len, user */
* catalog + 1 + 255 /* host_len, host */
*/ + 1 + 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) /* type, db_1, db_2, ... */
+ 1 + 4 /* + 1 + 3 /* type, microseconds */
* type, + 1 + 1 /* type, explicit_def..ts*/
* auto_increment + 1 + 8 /* type, xid of DDL */
*/ + 1 + 2 /* type, default_collation_for_utf8mb4_number */
+ 1 + 6 /* + 1 + 1 /* sql_require_primary_key */
* type, + 1 + 1 /* type, default_table_encryption */
* charset + 1 + 1 /* opt_flashback_area */
*/ + 1 + 1 /* type, opt_index_format_gpp_enabled */
+ 1 + 1 + 255 /* + 1 + 1 /* type, opt_index_format_panda_enabled */
* type, + 1 + 1 + NAME_LEN /* type, length, recycle_bin_table_name */);
* length
* ,
* time_zone
*/
+ 1 + 2 /*
* type,
* lc_time_names_number
*/
+ 1 + 2 /*
* type,
* charset_database_number
*/
+ 1 + 8 /*
* type,
* table_map_for_update
*/
+ 1 + 4 /*
* type,
* master_data_written
*/
/*
* type, db_1, db_2,
* ...
*/
/* type, microseconds */
/*
* MariaDb type,
* sec_part of NOW()
*/
+ 1 + (MAX_DBS_IN_EVENT_MTS * (1 + NAME_LEN)) + 3 /*
* type
* ,
* microseconds
*/+ 1 + 32
* 3 + 1 + 60/*
* type ,
* user_len
* , user ,
* host_len
* , host
*/)
+ 1 + 1 /*
* type,
* explicit_def
* ..ts
*/+ 1 + 8 /*
* type,
* xid
* of
* DDL
*/+ 1 + 2 /*
* type
* ,
* default_collation_for_utf8mb4_number
*/+ 1 /* sql_require_primary_key */;
/** /**
* Fixed data part: * Fixed data part:
* <ul> * <ul>
@@ -670,8 +617,75 @@ public class QueryLogEvent extends LogEvent {
public static final int Q_LIZARD_PREPARE_GCN = 201; public static final int Q_LIZARD_PREPARE_GCN = 201;
/**
* Replicate recycle_bin_table_name.
*/
public static final int Q_OPT_RECYCLE_BIN_TABLE_NAME = 254;
public static final int Q_OPT_INDEX_FORMAT_PANDA_ENABLED = 255; public static final int Q_OPT_INDEX_FORMAT_PANDA_ENABLED = 255;
private static final String findCodeName(final int code) {
switch (code) {
case Q_FLAGS2_CODE:
return "Q_FLAGS2_CODE";
case Q_SQL_MODE_CODE:
return "Q_SQL_MODE_CODE";
case Q_CATALOG_CODE:
return "Q_CATALOG_CODE";
case Q_AUTO_INCREMENT:
return "Q_AUTO_INCREMENT";
case Q_CHARSET_CODE:
return "Q_CHARSET_CODE";
case Q_TIME_ZONE_CODE:
return "Q_TIME_ZONE_CODE";
case Q_CATALOG_NZ_CODE:
return "Q_CATALOG_NZ_CODE";
case Q_LC_TIME_NAMES_CODE:
return "Q_LC_TIME_NAMES_CODE";
case Q_CHARSET_DATABASE_CODE:
return "Q_CHARSET_DATABASE_CODE";
case Q_TABLE_MAP_FOR_UPDATE_CODE:
return "Q_TABLE_MAP_FOR_UPDATE_CODE";
case Q_MASTER_DATA_WRITTEN_CODE:
return "Q_MASTER_DATA_WRITTEN_CODE";
case Q_INVOKER:
return "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_DEFAULT_TABLE_ENCRYPTION:
return "Q_DEFAULT_TABLE_ENCRYPTION";
case Q_OPT_FLASHBACK_AREA:
// or Q_DDL_SKIP_REWRITE
return "Q_OPT_FLASHBACK_AREA";
case Q_OPT_INDEX_FORMAT_GPP_ENABLED:
return "Q_OPT_INDEX_FORMAT_GPP_ENABLED";
case Q_OPT_RECYCLE_BIN_TABLE_NAME:
return "Q_OPT_RECYCLE_BIN_TABLE_NAME";
case Q_OPT_INDEX_FORMAT_PANDA_ENABLED:
return "Q_OPT_INDEX_FORMAT_PANDA_ENABLED";
case Q_HRNOW:
// or Q_WSREP_SKIP_READONLY_CHECKS
return "Q_HRNOW";
case Q_XID:
return "Q_XID";
case Q_GTID_FLAGS3:
return "Q_GTID_FLAGS3";
case Q_CHARACTER_SET_COLLATIONS :
return "Q_CHARACTER_SET_COLLATIONS";
}
return "CODE#" + code;
}
private final void unpackVariables(LogBuffer buffer, final int end) throws IOException { private final void unpackVariables(LogBuffer buffer, final int end) throws IOException {
int code = -1; int code = -1;
try { try {
@@ -830,6 +844,18 @@ public class QueryLogEvent extends LogEvent {
// prepareGCN = buffer.getLong64(); // prepareGCN = buffer.getLong64();
buffer.forward(8); buffer.forward(8);
break; break;
case Q_OPT_RECYCLE_BIN_TABLE_NAME:
int recycle_bin_table_name_len = buffer.getUint8();
String recycle_bin_table_name = null;
if (recycle_bin_table_name_len > 0) {
if (charset != null) {
recycle_bin_table_name = buffer.getFixLengthString(recycle_bin_table_name_len, charset);
} else {
recycle_bin_table_name = buffer.getFixLengthString(recycle_bin_table_name_len,
StandardCharsets.ISO_8859_1);
}
}
break;
case Q_OPT_INDEX_FORMAT_PANDA_ENABLED: case Q_OPT_INDEX_FORMAT_PANDA_ENABLED:
// *start++ = thd->variables.opt_index_format_panda_enabled; // *start++ = thd->variables.opt_index_format_panda_enabled;
buffer.forward(1); buffer.forward(1);
@@ -849,66 +875,6 @@ public class QueryLogEvent extends LogEvent {
} }
} }
private static final String findCodeName(final int code) {
switch (code) {
case Q_FLAGS2_CODE:
return "Q_FLAGS2_CODE";
case Q_SQL_MODE_CODE:
return "Q_SQL_MODE_CODE";
case Q_CATALOG_CODE:
return "Q_CATALOG_CODE";
case Q_AUTO_INCREMENT:
return "Q_AUTO_INCREMENT";
case Q_CHARSET_CODE:
return "Q_CHARSET_CODE";
case Q_TIME_ZONE_CODE:
return "Q_TIME_ZONE_CODE";
case Q_CATALOG_NZ_CODE:
return "Q_CATALOG_NZ_CODE";
case Q_LC_TIME_NAMES_CODE:
return "Q_LC_TIME_NAMES_CODE";
case Q_CHARSET_DATABASE_CODE:
return "Q_CHARSET_DATABASE_CODE";
case Q_TABLE_MAP_FOR_UPDATE_CODE:
return "Q_TABLE_MAP_FOR_UPDATE_CODE";
case Q_MASTER_DATA_WRITTEN_CODE:
return "Q_MASTER_DATA_WRITTEN_CODE";
case Q_INVOKER:
return "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_DEFAULT_TABLE_ENCRYPTION:
return "Q_DEFAULT_TABLE_ENCRYPTION";
case Q_OPT_FLASHBACK_AREA:
// or Q_DDL_SKIP_REWRITE
return "Q_OPT_FLASHBACK_AREA";
case Q_OPT_INDEX_FORMAT_GPP_ENABLED:
return "Q_OPT_INDEX_FORMAT_GPP_ENABLED";
case Q_OPT_INDEX_FORMAT_PANDA_ENABLED:
return "Q_OPT_INDEX_FORMAT_PANDA_ENABLED";
case Q_HRNOW:
// or Q_WSREP_SKIP_READONLY_CHECKS
return "Q_HRNOW";
case Q_XID:
return "Q_XID";
case Q_GTID_FLAGS3:
return "Q_GTID_FLAGS3";
case Q_CHARACTER_SET_COLLATIONS :
return "Q_CHARACTER_SET_COLLATIONS";
}
return "CODE#" + code;
}
public final String getUser() { public final String getUser() {
return user; return user;
} }