Unlocking MySQL Binlog: How Replication, Recovery, and Auditing Really Work

MySQL Master‑Slave Working Mechanism

What is binlog?

binlog records all database table structure changes (e.g., CREATE, ALTER TABLE) and table data modifications (INSERT, UPDATE, DELETE) in a binary log.

binlog does not record SELECT and SHOW because they do not modify data; you can view them via the general log. Even UPDATE statements that do not change data are still recorded in binlog.

Uses of binlog

Reference "MySQL Technical Internals InnoDB Storage Engine" for uses.

Recovery: point‑in‑time recovery using binlog after a full backup.

Replication: master sends binlog to slave for real‑time synchronization.

Audit: use binlog information to audit and detect possible injection attacks.

Role of binlog in master‑slave synchronization

Master‑slave sync principle

Master‑slave sync process

There are synchronous and asynchronous replication; most real‑world setups use asynchronous replication.

Basic replication steps:

Slave IO thread connects to Master and requests binlog from a specific file/position.

Master receives the request, reads the binlog from that position, returns the data and updates the master‑info file with the binlog file name and position.

Slave IO thread appends the received binlog to its relay‑log file and records the master binlog file name and position.

Slave SQL thread parses new relay‑log entries and executes them on the slave.

Canal Working Mechanism

Canal synchronizes MySQL data based on binlog. It mimics the MySQL master‑slave protocol.

Working principle

Canal simulates a MySQL slave and sends a dump request to the MySQL master.

Master receives the dump request and pushes the binary log to the slave (Canal).

Canal parses the binary log object (byte stream).

Binlog parsing process

Destination startup

During CanalServer startup, each destination is started via the Alibaba Otter Canal controller.

public void processActiveEnter() {
    try {
        MDC.put(CanalConstants.MDC_DESTINATION, String.valueOf(destination));
        embededCanalServer.start(destination);
        if (canalMQStarter != null) {
            canalMQStarter.startDestination(destination);
        }
    } finally {
        MDC.remove(CanalConstants.MDC_DESTINATION);
    }
}

The embedded start method actually starts the instance:

public void start(final String destination) {
    final CanalInstance canalInstance = canalInstances.get(destination);
    if (!canalInstance.isStart()) {
        try {
            MDC.put("destination", destination);
            if (metrics.isRunning()) {
                metrics.register(canalInstance);
            }
            canalInstance.start();
            logger.info("start CanalInstances[{}] successfully", destination);
        } finally {
            MDC.remove("destination");
        }
    }
}

CanalInstance class hierarchy is shown (image).

Replication component processes MySQL dump commands, registers slave info, and receives binlog.

Binlog parsing is implemented by com.alibaba.otter.canal.parse.inbound.BinlogParser.

LogEvent defines binary log events; LogEventConvert converts them.

@Override
public Entry parse(LogEvent logEvent, boolean isSeek) throws CanalParseException {
    if (logEvent == null || logEvent instanceof UnknownLogEvent) {
        return null;
    }
    int eventType = logEvent.getHeader().getType();
    switch (eventType) {
        case LogEvent.QUERY_EVENT:
            return parseQueryEvent((QueryLogEvent) logEvent, isSeek);
        case LogEvent.XID_EVENT:
            return parseXidEvent((XidLogEvent) logEvent);
        case LogEvent.TABLE_MAP_EVENT:
            break;
        case LogEvent.WRITE_ROWS_EVENT_V1:
        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:
        case LogEvent.DELETE_ROWS_EVENT:
            return parseRowsEvent((DeleteRowsLogEvent) logEvent);
        case LogEvent.ROWS_QUERY_LOG_EVENT:
            return parseRowsQueryEvent((RowsQueryLogEvent) logEvent);
        case LogEvent.ANNOTATE_ROWS_EVENT:
            return parseAnnotateRowsEvent((AnnotateRowsEvent) logEvent);
        case LogEvent.USER_VAR_EVENT:
            return parseUserVarLogEvent((UserVarLogEvent) logEvent);
        case LogEvent.INTVAR_EVENT:
            return parseIntrvarLogEvent((IntvarLogEvent) logEvent);
        case LogEvent.RAND_EVENT:
            return parseRandLogEvent((RandLogEvent) logEvent);
        case LogEvent.GTID_LOG_EVENT:
            return parseGTIDLogEvent((GtidLogEvent) logEvent);
        case LogEvent.HEARTBEAT_LOG_EVENT:
            return parseHeartbeatLogEvent((HeartbeatLogEvent) logEvent);
        default:
            break;
    }
    return null;
}

Row parsing core:

while (buffer.nextOneRow(columns, false)) {
    // process row record
    RowData.Builder rowDataBuilder = RowData.newBuilder();
    if (EventType.INSERT == eventType) {
        // insert records go to before field
        tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, true, tableMeta);
    } else if (EventType.DELETE == eventType) {
        // delete records go to before field
        tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
    } else {
        // update needs before/after
        tableError |= parseOneRow(rowDataBuilder, event, buffer, columns, false, tableMeta);
        if (!buffer.nextOneRow(changeColumns, true)) {
            rowChangeBuider.addRowDatas(rowDataBuilder.build());
            break;
        }
        tableError |= parseOneRow(rowDataBuilder, event, buffer, changeColumns, true, tableMeta);
    }
    rowsCount++;
    rowChangeBuider.addRowDatas(rowDataBuilder.build());
}

Parsing individual column values:

switch (javaType) {
    case Types.INTEGER:
    case Types.TINYINT:
    case Types.SMALLINT:
    case Types.BIGINT:
        // handle unsigned types
        Number number = (Number) value;
        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 + number.intValue())));
                    javaType = Types.SMALLINT;
                    break;
                case 2: /* MYSQL_TYPE_SHORT */
                    columnBuilder.setValue(String.valueOf(Integer.valueOf(SMALLINT_MAX_VALUE + number.intValue())));
                    javaType = Types.INTEGER;
                    break;
                case 3: /* MYSQL_TYPE_INT24 */
                    columnBuilder.setValue(String.valueOf(Integer.valueOf(MEDIUMINT_MAX_VALUE + number.intValue())));
                    javaType = Types.INTEGER;
                    break;
                case 4: /* MYSQL_TYPE_LONG */
                    columnBuilder.setValue(String.valueOf(Long.valueOf(INTEGER_MAX_VALUE + number.longValue())));
                    javaType = Types.BIGINT;
                    break;
                case 8: /* MYSQL_TYPE_LONGLONG */
                    columnBuilder.setValue(BIGINT_MAX_VALUE.add(BigInteger.valueOf(number.longValue())).toString());
                    javaType = Types.DECIMAL;
                    break;
            }
        } else {
            // number type, direct valueOf
            columnBuilder.setValue(String.valueOf(value));
        }
        if (isSingleBit && javaType == Types.TINYINT) {
            javaType = Types.BIT;
        }
        break;
    case Types.REAL: // float
    case Types.DOUBLE: // double
        columnBuilder.setValue(String.valueOf(value));
        break;
    case Types.BIT: // bit
        columnBuilder.setValue(String.valueOf(value));
        break;
    case Types.DECIMAL:
        columnBuilder.setValue(((BigDecimal) value).toPlainString());
        break;
    case Types.TIMESTAMP:
        // timestamp handling omitted for brevity
        break;
    case Types.TIME:
    case Types.DATE:
        columnBuilder.setValue(value.toString());
        break;
    case Types.BINARY:
    case Types.VARBINARY:
    case Types.LONGVARBINARY:
        if (fieldMeta != null && isText(fieldMeta.getColumnType())) {
            columnBuilder.setValue(new String((byte[]) value, charset));
            javaType = Types.CLOB;
        } else {
            columnBuilder.setValue(new String((byte[]) value, ISO_8859_1));
            javaType = Types.BLOB;
        }
        break;
    case Types.CHAR:
    case Types.VARCHAR:
        columnBuilder.setValue(value.toString());
        break;
    default:
        columnBuilder.setValue(value.toString());
}

The final output is the familiar data structure seen by consumers.