diff --git a/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java b/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java index b9e88b5..e6c884a 100644 --- a/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java +++ b/src/main/java/com/github/shyiko/mysql/binlog/BinaryLogClient.java @@ -30,6 +30,11 @@ import com.github.shyiko.mysql.binlog.event.deserialization.GtidEventDataDeserializer; import com.github.shyiko.mysql.binlog.event.deserialization.QueryEventDataDeserializer; import com.github.shyiko.mysql.binlog.event.deserialization.RotateEventDataDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.maria.BinlogCheckpointDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.maria.GtidDeserializer; +import com.github.shyiko.mysql.binlog.event.deserialization.maria.GtidListDeserializer; +import com.github.shyiko.mysql.binlog.event.maria.Gtid; +import com.github.shyiko.mysql.binlog.event.maria.MariaGtidEventData; import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; import com.github.shyiko.mysql.binlog.jmx.BinaryLogClientMXBean; import com.github.shyiko.mysql.binlog.network.AuthenticationException; @@ -46,13 +51,7 @@ import com.github.shyiko.mysql.binlog.network.protocol.Packet; import com.github.shyiko.mysql.binlog.network.protocol.PacketChannel; import com.github.shyiko.mysql.binlog.network.protocol.ResultSetRowPacket; -import com.github.shyiko.mysql.binlog.network.protocol.command.AuthenticateCommand; -import com.github.shyiko.mysql.binlog.network.protocol.command.Command; -import com.github.shyiko.mysql.binlog.network.protocol.command.DumpBinaryLogCommand; -import com.github.shyiko.mysql.binlog.network.protocol.command.DumpBinaryLogGtidCommand; -import com.github.shyiko.mysql.binlog.network.protocol.command.PingCommand; -import com.github.shyiko.mysql.binlog.network.protocol.command.QueryCommand; -import com.github.shyiko.mysql.binlog.network.protocol.command.SSLRequestCommand; +import com.github.shyiko.mysql.binlog.network.protocol.command.*; import javax.net.ssl.SSLContext; import javax.net.ssl.TrustManager; @@ -158,6 +157,11 @@ public X509Certificate[] getAcceptedIssuers() { private Event previousEvent; private Event previousGtidEvent; + // MariaDB + private Gtid mariaGtid; + private String gtid; + private boolean mariaDB; + /** * Alias for BinaryLogClient("localhost", 3306, <no schema> = null, username, password). * @see BinaryLogClient#BinaryLogClient(String, int, String, String, String) @@ -834,6 +838,9 @@ private void updatePosition(Event event, boolean updatedGtid) { if (nextBinlogPosition > 0) { binlogPosition = nextBinlogPosition; } + } else + if (eventType == EventType.MARIA_GTID_EVENT){ + updateMariaGTID(event); } previousEvent = event; } @@ -1020,6 +1027,101 @@ private void disconnectChannel() throws IOException { } } + /** + * @return Note that this value changes with each received GTID event (provided client is in GTID mode). + */ + public String getGtid() { + synchronized (gtidSetAccessLock) { + if (gtidSet != null) { + return gtidSet.toString(); + } + return gtid; + } + } + + /** + * @param gtid For MySQL this is GTID set format, for MariaDB the format is domainId-serverId-sequenceNumber(can be an empty string). + *

NOTE #1: Any value but null will switch BinaryLogClient into a GTID mode (in which case GTID set will be + * updated with each incoming GTID event) as well as set binlogFilename to "" (empty string) (meaning + * BinaryLogClient will request events "outside of the set" starting from the oldest known binlog). + *

NOTE #2: {@link #setBinlogFilename(String)} and {@link #setBinlogPosition(long)} can be used to specify the + * exact position from which MySQL server should start streaming events (taking into account GTID set). + * @see #getGtid() + */ + public BinaryLogClient setGtid(String gtid) { + if (gtid != null && this.binlogFilename == null) { + this.binlogFilename = ""; + } + synchronized (gtidSetAccessLock) { + this.gtid = gtid; + } + return this; + } + + // region MariaDB + public boolean isMariaDB() { + return mariaDB; + } + + private void updateMariaGTID(Event event) { + EventHeader eventHeader = event.getHeader(); + if (eventHeader.getEventType() == EventType.MARIA_GTID_EVENT) { + synchronized (gtidSetAccessLock) { + if (mariaGtid != null) { + MariaGtidEventData eventData = event.getData(); + mariaGtid.setDomainId(eventData.getDomainId()); + mariaGtid.setSequenceNumber(eventData.getSequenceNumber()); + gtid = mariaGtid.toString(); + } + } + } + } + + private DumpBinaryLogCommand requestMariaBinaryLogStream() throws IOException { + if ("gtid_current_pos".equals(gtid) || "".equals(gtid)||gtid==null) { + channel.write(new QueryCommand("select @@gtid_current_pos")); + ResultSetRowPacket[] rs = readResultSet(); + gtid = rs[0].getValue(0); + logger.fine("Use server current gtid position "+gtid); + } + + // update server id + channel.write(new QueryCommand("SHOW VARIABLES LIKE 'SERVER_ID'")); + ResultSetRowPacket[] rs = readResultSet(); + long serverId = Long.parseLong(rs[0].getValue(1)); + // If we got multi gtid, chose the gtid for current server + String[] split = gtid.split(","); + for (String s : split) { + Gtid g = new Gtid(s); + if (g.getServerId() == serverId) { + mariaGtid = g; + gtid = mariaGtid.toString(); + logger.fine("Chose gtid "+gtid+" for this server"); + } + } + + // set up gtid + channel.write(new QueryCommand("SET @mariadb_slave_capability = 4"));// support GTID + channel.read();// ignore + channel.write(new QueryCommand("SET @slave_connect_state = '" + gtid + "'")); + channel.read();// ignore + channel.write(new QueryCommand("SET @slave_gtid_strict_mode = 0")); + channel.read();// ignore + channel.write(new QueryCommand("SET @slave_gtid_ignore_duplicates = 0")); + channel.read();// ignore + // Register First + Command command = new RegisterSlaveCommand(serverId, "", "", "", 0, 0, 0); + channel.write(command); + channel.read();// ignore + + // MariaDB Event + eventDeserializer.setEventDataDeserializer(EventType.MARIA_GTID_EVENT, new GtidDeserializer()); + eventDeserializer.setEventDataDeserializer(EventType.MARIA_GTID_LIST_EVENT, new GtidListDeserializer()); + eventDeserializer.setEventDataDeserializer(EventType.MARIA_BINLOG_CHECKPOINT_EVENT, new BinlogCheckpointDeserializer()); + + return new DumpBinaryLogCommand(serverId, binlogFilename, binlogPosition); + } + // endregion /** * {@link BinaryLogClient}'s event listener. */ diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/EventType.java b/src/main/java/com/github/shyiko/mysql/binlog/event/EventType.java index 6d1adbb..a9db10e 100644 --- a/src/main/java/com/github/shyiko/mysql/binlog/event/EventType.java +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/EventType.java @@ -186,7 +186,49 @@ public enum EventType { */ GTID, ANONYMOUS_GTID, - PREVIOUS_GTIDS; + PREVIOUS_GTIDS, + + /* MARIA_EVENT_BEGIN */ + + MARIA_ANNOTATE_ROWS_EVENT(160), + /** + * Binlog checkpoint event. Used for XA crash recovery on the master, not used + * in replication. + * A binlog checkpoint event specifies a binlog file such that XA crash + * recovery can start from that file - and it is guaranteed to find all XIDs + * that are prepared in storage engines but not yet committed. + */ + MARIA_BINLOG_CHECKPOINT_EVENT(161), + /** + * Gtid event. For global transaction ID, used to start a new event group, + * instead of the old BEGIN query event, and also to mark stand-alone + * events. + */ + MARIA_GTID_EVENT(162), + /** + * Gtid list event. Logged at the start of every binlog, to record the + * current replication state. This consists of the last GTID seen for + * each replication domain. + */ + MARIA_GTID_LIST_EVENT(163), + ; + + final int v; + + EventType() + { + v = ordinal(); + } + + EventType(int n) + { + v = n; + } + + public int get() + { + return v; + } public static boolean isRowMutation(EventType eventType) { return EventType.isWrite(eventType) || diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java index 474f05d..11acf4d 100644 --- a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventDeserializer.java @@ -33,9 +33,9 @@ */ public class EventDeserializer { - private final EventHeaderDeserializer eventHeaderDeserializer; - private final EventDataDeserializer defaultEventDataDeserializer; - private final Map eventDataDeserializers; + protected final EventHeaderDeserializer eventHeaderDeserializer; + protected final EventDataDeserializer defaultEventDataDeserializer; + protected final Map eventDataDeserializers; private int checksumLength; diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventHeaderV4Deserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventHeaderV4Deserializer.java index 2d8bcdd..2d599fd 100644 --- a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventHeaderV4Deserializer.java +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/EventHeaderV4Deserializer.java @@ -20,14 +20,25 @@ import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; import java.io.IOException; +import java.util.HashMap; +import java.util.Map; /** * @author Stanley Shyiko */ public class EventHeaderV4Deserializer implements EventHeaderDeserializer { - private static final EventType[] EVENT_TYPES = EventType.values(); + private static final Map EVENT_TYPES; + + static { + // Mutable map but will never update, so it's safe. + EVENT_TYPES = new HashMap(); + for (EventType type : EventType.values()) + { + EVENT_TYPES.put(type.get(),type); + } + } @Override public EventHeaderV4 deserialize(ByteArrayInputStream inputStream) throws IOException { EventHeaderV4 header = new EventHeaderV4(); @@ -41,10 +52,11 @@ public EventHeaderV4 deserialize(ByteArrayInputStream inputStream) throws IOExce } private static EventType getEventType(int ordinal) throws IOException { - if (ordinal >= EVENT_TYPES.length) { + EventType type = EVENT_TYPES.get(ordinal); + if (type == null) { throw new IOException("Unknown event type " + ordinal); } - return EVENT_TYPES[ordinal]; + return type; } } diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/BinlogCheckpointDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/BinlogCheckpointDeserializer.java new file mode 100644 index 0000000..01543e8 --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/BinlogCheckpointDeserializer.java @@ -0,0 +1,19 @@ +package com.github.shyiko.mysql.binlog.event.deserialization.maria; + +import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializer; +import com.github.shyiko.mysql.binlog.event.maria.BinlogCheckpointEventData; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; + +import java.io.IOException; + +/** + * @author wener + */ +public class BinlogCheckpointDeserializer implements EventDataDeserializer { + @Override + public BinlogCheckpointEventData deserialize(ByteArrayInputStream is) throws IOException { + BinlogCheckpointEventData e = new BinlogCheckpointEventData(); + e.setBinlogFilename(is.readString(is.readInteger(4))); + return e; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidDeserializer.java new file mode 100644 index 0000000..0ea8d89 --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidDeserializer.java @@ -0,0 +1,26 @@ +package com.github.shyiko.mysql.binlog.event.deserialization.maria; + +import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializer; +import com.github.shyiko.mysql.binlog.event.maria.MariaGtidEventData; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; + +import java.io.IOException; + +/** + * @author wener + */ +public class GtidDeserializer implements EventDataDeserializer { + @Override + public MariaGtidEventData deserialize(ByteArrayInputStream is) throws IOException { + MariaGtidEventData e = new MariaGtidEventData(); + e.setSequenceNumber(is.readLong(8)); + e.setDomainId(is.readLong(4)); + e.setFlags(is.readInteger(1)); + + // reserved + long n = 6 + ((e.getFlags() & MariaGtidEventData.FL_GROUP_COMMIT_ID) > 0 ? 2 : 0); + long skip = is.skip(n); + assert n == skip; + return e; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidListDeserializer.java b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidListDeserializer.java new file mode 100644 index 0000000..98785ab --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/deserialization/maria/GtidListDeserializer.java @@ -0,0 +1,27 @@ +package com.github.shyiko.mysql.binlog.event.deserialization.maria; + +import com.github.shyiko.mysql.binlog.event.deserialization.EventDataDeserializer; +import com.github.shyiko.mysql.binlog.event.maria.Gtid; +import com.github.shyiko.mysql.binlog.event.maria.GtidListEventData; +import com.github.shyiko.mysql.binlog.io.ByteArrayInputStream; + +import java.io.IOException; + +/** + * @author wener + */ +public class GtidListDeserializer implements EventDataDeserializer { + @Override + public GtidListEventData deserialize(ByteArrayInputStream is) throws IOException { + GtidListEventData e = new GtidListEventData(); + e.getList().clear(); + long count = is.readLong(4); + e.setFlag((int) (count >> 28));// higher 4 bit + count = count & 0x1fffffff;// lower 28 bit + + for (long i = 0; i < count; i++) { + e.getList().add(new Gtid(is.readLong(4), is.readLong(4), is.readLong(8))); + } + return e; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/maria/BinlogCheckpointEventData.java b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/BinlogCheckpointEventData.java new file mode 100644 index 0000000..9ec2f89 --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/BinlogCheckpointEventData.java @@ -0,0 +1,26 @@ +package com.github.shyiko.mysql.binlog.event.maria; + +import com.github.shyiko.mysql.binlog.event.EventData; + +/** + * @author wener + */ +public class BinlogCheckpointEventData implements EventData { + private String binlogFilename; + + public String getBinlogFilename() { + return binlogFilename; + } + + public BinlogCheckpointEventData setBinlogFilename(String binlogFilename) { + this.binlogFilename = binlogFilename; + return this; + } + + @Override + public String toString() { + return "BinlogCheckpointEventData{" + + "binlogFilename='" + binlogFilename + '\'' + + '}'; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/maria/Gtid.java b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/Gtid.java new file mode 100644 index 0000000..4440b9b --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/Gtid.java @@ -0,0 +1,57 @@ +package com.github.shyiko.mysql.binlog.event.maria; + +/** + * @author wener + */ +public class Gtid { + private long domainId; + private long serverId; + private long sequenceNumber;//FIXME unsigned long + + public Gtid(long domainId, long serverId, long sequenceNumber) { + this.domainId = domainId; + this.serverId = serverId; + this.sequenceNumber = sequenceNumber; + } + + public Gtid(String gtid) { + if (gtid != null && !gtid.isEmpty()) { + String[] split = gtid.split("-"); + domainId = Long.parseLong(split[0]); + serverId = Long.parseLong(split[1]); + sequenceNumber = Long.parseLong(split[2]); + } + } + + public long getDomainId() { + return domainId; + } + + public Gtid setDomainId(long domainId) { + this.domainId = domainId; + return this; + } + + public long getServerId() { + return serverId; + } + + public Gtid setServerId(long serverId) { + this.serverId = serverId; + return this; + } + + public long getSequenceNumber() { + return sequenceNumber; + } + + public Gtid setSequenceNumber(long sequenceNumber) { + this.sequenceNumber = sequenceNumber; + return this; + } + + @Override + public String toString() { + return String.format("%s-%s-%s", domainId, serverId, sequenceNumber); + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/maria/GtidListEventData.java b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/GtidListEventData.java new file mode 100644 index 0000000..ab5a043 --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/GtidListEventData.java @@ -0,0 +1,36 @@ +package com.github.shyiko.mysql.binlog.event.maria; + +import com.github.shyiko.mysql.binlog.event.EventData; + +import java.util.ArrayList; +import java.util.List; + +/** + * @author wener + */ +public class GtidListEventData implements EventData { + // never null + private final List list = new ArrayList(); + private int flag; + + public List getList() { + return list; + } + + public int getFlag() { + return flag; + } + + public GtidListEventData setFlag(int flag) { + this.flag = flag; + return this; + } + + @Override + public String toString() { + return "GtidListEventData{" + + "list=" + list + + ", flag=" + flag + + '}'; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/event/maria/MariaGtidEventData.java b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/MariaGtidEventData.java new file mode 100644 index 0000000..3f15b0c --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/event/maria/MariaGtidEventData.java @@ -0,0 +1,74 @@ +package com.github.shyiko.mysql.binlog.event.maria; + +import com.github.shyiko.mysql.binlog.event.EventData; + +public class MariaGtidEventData implements EventData { + /** + * FL_STANDALONE is set when there is no terminating COMMIT event. + */ + public static final int FL_STANDALONE = 1; + /** + * FL_GROUP_COMMIT_ID is set when event group is part of a group commit on the + * master. Groups with same commit_id are part of the same group commit. + */ + public static final int FL_GROUP_COMMIT_ID = 2; + /** + * FL_TRANSACTIONAL is set for an event group that can be safely rolled back + * (no MyISAM, eg.). + */ + public static final int FL_TRANSACTIONAL = 4; + + /* Flags. */ + /** + * FL_ALLOW_PARALLEL reflects the (negation of the) value of @@SESSION.skip_parallel_replication at the time of commit. + */ + public static final int FL_ALLOW_PARALLEL = 8; + /** + * FL_WAITED is set if a row lock wait (or other wait) is detected during the + * execution of the transaction. + */ + public static final int FL_WAITED = 16; + /** + * FL_DDL is set for event group containing DDL. + */ + public static final int FL_DDL = 32; + private long sequenceNumber;//8 + private long domainId;// 4 + private int flags;// 1 + + public long getSequenceNumber() { + return sequenceNumber; + } + + public MariaGtidEventData setSequenceNumber(long sequenceNumber) { + this.sequenceNumber = sequenceNumber; + return this; + } + + public long getDomainId() { + return domainId; + } + + public MariaGtidEventData setDomainId(long domainId) { + this.domainId = domainId; + return this; + } + + public int getFlags() { + return flags; + } + + public MariaGtidEventData setFlags(int flags) { + this.flags = flags; + return this; + } + + @Override + public String toString() { + return "GtidEventData{" + + "sequenceNumber=" + sequenceNumber + + ", domainId=" + domainId + + ", flags2=" + flags + + '}'; + } +} diff --git a/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/RegisterSlaveCommand.java b/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/RegisterSlaveCommand.java new file mode 100644 index 0000000..63cd5b2 --- /dev/null +++ b/src/main/java/com/github/shyiko/mysql/binlog/network/protocol/command/RegisterSlaveCommand.java @@ -0,0 +1,45 @@ +package com.github.shyiko.mysql.binlog.network.protocol.command; + +import com.github.shyiko.mysql.binlog.io.ByteArrayOutputStream; + +import java.io.IOException; + +/** + * @author wener + */ +public class RegisterSlaveCommand implements Command { + private long serverId; + private String slaveHostname; + private String slaveUser; + private String slavePassword; + private int slavePort; + private long replicationRank; + private long masterId; + + public RegisterSlaveCommand(long serverId, String slaveHostname, String slaveUser, String slavePassword, int slavePort, long replicationRank, long masterId) { + this.serverId = serverId; + this.slaveHostname = slaveHostname; + this.slaveUser = slaveUser; + this.slavePassword = slavePassword; + this.slavePort = slavePort; + this.replicationRank = replicationRank; + this.masterId = masterId; + } + + @Override + public byte[] toByteArray() throws IOException { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(); + buffer.write(CommandType.REGISTER_SLAVE.ordinal()); + buffer.writeLong(serverId, 4); + buffer.writeInteger(slaveHostname.length(), 1); + buffer.writeString(slaveHostname); + buffer.writeInteger(slaveUser.length(), 1); + buffer.writeString(slaveUser); + buffer.writeInteger(slavePassword.length(), 1); + buffer.writeString(slavePassword); + buffer.writeInteger(slavePort, 2); + buffer.writeLong(replicationRank, 4); + buffer.writeLong(masterId, 4); + return buffer.toByteArray(); + } +}