You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@shardingsphere.apache.org by du...@apache.org on 2022/12/23 07:15:45 UTC
[shardingsphere] branch master updated: Refactor MySQL sequence ID handling logic (#23053)
This is an automated email from the ASF dual-hosted git repository.
duanzhengqiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shardingsphere.git
The following commit(s) were added to refs/heads/master by this push:
new 7eef6b2b52d Refactor MySQL sequence ID handling logic (#23053)
7eef6b2b52d is described below
commit 7eef6b2b52d37453fec69b38385baddeef85dc55
Author: 吴伟杰 <wu...@apache.org>
AuthorDate: Fri Dec 23 15:15:34 2022 +0800
Refactor MySQL sequence ID handling logic (#23053)
* Refactor MySQL sequence ID handling logic
* Fix sequence ID when writing multi packets
* Fix decoding multi packets
* Complete tests in shardingsphere-mysql-protocol
* Complete tests in shardingsphere-proxy-frontend-mysql
* Initial sequence ID in MySQLClient
* Remove sequence ID from AbstractMySQLBinlogEventPacket
* Remove sequence ID from MySQLBinlogEventPacketDecoder
* Complete tests in shardingsphere-data-pipeline-mysql
* Fix getByte offset in MySQLCommandPacketDecoder
* Fix getByte offset in MySQLNegotiatePackageDecoder
* Fix magic value in MySQLBinlogEventPacketDecoder
---
.../mysql/codec/MySQLPacketCodecEngine.java | 24 +++++++++++++---------
.../db/protocol/mysql/constant/MySQLConstants.java | 4 ++++
.../db/protocol/mysql/packet/MySQLPacket.java | 7 -------
.../binlog/AbstractMySQLBinlogEventPacket.java | 9 ++------
.../packet/binlog/MySQLBinlogEventHeader.java | 5 -----
.../mysql/packet/command/MySQLCommandPacket.java | 5 -----
.../command/MySQLCommandPacketTypeLoader.java | 2 --
.../query/MySQLColumnDefinition41Packet.java | 16 +++++----------
.../command/query/MySQLFieldCountPacket.java | 4 +---
.../execute/MySQLBinaryResultSetRowPacket.java | 4 ----
.../prepare/MySQLComStmtPrepareOKPacket.java | 4 ----
.../query/text/MySQLTextResultSetRowPacket.java | 3 ---
.../mysql/packet/generic/MySQLEofPacket.java | 7 ++-----
.../mysql/packet/generic/MySQLErrPacket.java | 13 ++++--------
.../mysql/packet/generic/MySQLOKPacket.java | 11 ++++------
.../packet/handshake/MySQLAuthMoreDataPacket.java | 4 ----
.../handshake/MySQLAuthSwitchRequestPacket.java | 4 ----
.../handshake/MySQLAuthSwitchResponsePacket.java | 4 ----
.../packet/handshake/MySQLHandshakePacket.java | 6 ------
.../handshake/MySQLHandshakeResponse41Packet.java | 3 ---
.../mysql/codec/MySQLPacketCodecEngineTest.java | 10 +++++----
.../packet/binlog/MySQLBinlogEventHeaderTest.java | 1 -
...ySQLBinlogFormatDescriptionEventPacketTest.java | 1 -
.../MySQLBinlogRotateEventPacketTest.java | 1 -
.../row/MySQLBinlogTableMapEventPacketTest.java | 1 +
.../command/MySQLCommandPacketTypeLoaderTest.java | 4 ++--
.../admin/initdb/MySQLComInitDbPacketTest.java | 1 -
.../query/MySQLColumnDefinition41PacketTest.java | 8 ++------
.../command/query/MySQLFieldCountPacketTest.java | 3 +--
.../execute/MySQLBinaryResultSetRowPacketTest.java | 5 +----
.../MySQLComStmtPrepareMySQLOKPacketTest.java | 5 +----
.../prepare/MySQLComStmtPreparePacketTest.java | 1 -
.../text/MySQLTextResultSetRowPacketTest.java | 10 +++------
.../query/text/query/MySQLComQueryPacketTest.java | 1 -
.../mysql/packet/generic/MySQLEofPacketTest.java | 10 ++++-----
.../mysql/packet/generic/MySQLErrPacketTest.java | 8 +++-----
.../mysql/packet/generic/MySQLOKPacketTest.java | 13 +++++-------
.../MySQLAuthSwitchRequestPacketTest.java | 2 +-
.../MySQLAuthSwitchResponsePacketTest.java | 2 --
.../packet/handshake/MySQLHandshakePacketTest.java | 6 ++----
.../MySQLHandshakeResponse41PacketTest.java | 22 ++++++++------------
.../pipeline/mysql/ingest/client/MySQLClient.java | 11 ++++++++++
.../netty/MySQLBinlogEventPacketDecoder.java | 11 +++-------
.../client/netty/MySQLCommandPacketDecoder.java | 6 +++---
.../ingest/client/netty/MySQLNegotiateHandler.java | 8 ++++----
.../client/netty/MySQLNegotiatePackageDecoder.java | 2 +-
.../mysql/ingest/client/MySQLClientTest.java | 13 +++++++++---
.../netty/MySQLBinlogEventPacketDecoderTest.java | 18 ++++++++--------
.../netty/MySQLCommandPacketDecoderTest.java | 16 +++++++--------
.../client/netty/MySQLNegotiateHandlerTest.java | 4 ++--
.../netty/MySQLNegotiatePackageDecoderTest.java | 10 ++++-----
.../authentication/MySQLAuthenticationEngine.java | 16 +++++++--------
.../mysql/command/MySQLCommandExecuteEngine.java | 4 +---
.../admin/MySQLComResetConnectionExecutor.java | 2 +-
.../command/admin/MySQLComSetOptionExecutor.java | 2 +-
.../admin/initdb/MySQLComInitDbExecutor.java | 2 +-
.../command/admin/ping/MySQLComPingExecutor.java | 2 +-
.../command/admin/quit/MySQLComQuitExecutor.java | 2 +-
.../execute/MySQLComStmtExecuteExecutor.java | 8 ++------
.../prepare/MySQLComStmtPrepareExecutor.java | 12 +++++------
.../binary/reset/MySQLComStmtResetExecutor.java | 2 +-
.../query/builder/ResponsePacketBuilder.java | 9 ++++----
.../fieldlist/MySQLComFieldListPacketExecutor.java | 6 ++----
.../text/query/MySQLComQueryPacketExecutor.java | 8 ++------
.../frontend/mysql/err/MySQLErrPacketFactory.java | 4 ++--
.../frontend/mysql/MySQLFrontendEngineTest.java | 1 +
.../MySQLAuthenticationEngineTest.java | 1 +
.../execute/MySQLComStmtExecuteExecutorTest.java | 1 -
.../mysql/err/MySQLErrPacketFactoryTest.java | 5 -----
69 files changed, 166 insertions(+), 274 deletions(-)
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngine.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngine.java
index a08ff36de8a..b1e50337054 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngine.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngine.java
@@ -22,6 +22,7 @@ import io.netty.buffer.CompositeByteBuf;
import io.netty.channel.ChannelHandlerContext;
import org.apache.shardingsphere.db.protocol.CommonConstants;
import org.apache.shardingsphere.db.protocol.codec.DatabasePacketCodecEngine;
+import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLConstants;
import org.apache.shardingsphere.db.protocol.mysql.packet.MySQLPacket;
import org.apache.shardingsphere.db.protocol.mysql.packet.generic.MySQLErrPacket;
import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@@ -32,6 +33,7 @@ import java.sql.SQLException;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Database packet codec for MySQL.
@@ -59,7 +61,9 @@ public final class MySQLPacketCodecEngine implements DatabasePacketCodecEngine<M
in.resetReaderIndex();
return;
}
- ByteBuf message = in.readRetainedSlice(SEQUENCE_LENGTH + payloadLength);
+ short sequenceId = in.readUnsignedByte();
+ context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get().set(sequenceId + 1);
+ ByteBuf message = in.readRetainedSlice(payloadLength);
if (MAX_PACKET_LENGTH == payloadLength) {
pendingMessages.add(message);
} else if (pendingMessages.isEmpty()) {
@@ -74,10 +78,10 @@ public final class MySQLPacketCodecEngine implements DatabasePacketCodecEngine<M
Iterator<ByteBuf> pendingMessagesIterator = pendingMessages.iterator();
result.addComponent(true, pendingMessagesIterator.next());
while (pendingMessagesIterator.hasNext()) {
- result.addComponent(true, pendingMessagesIterator.next().skipBytes(1));
+ result.addComponent(true, pendingMessagesIterator.next());
}
- if (lastMessage.readableBytes() > 1) {
- result.addComponent(true, lastMessage.skipBytes(1));
+ if (lastMessage.readableBytes() > 0) {
+ result.addComponent(true, lastMessage);
}
out.add(result);
pendingMessages.clear();
@@ -93,13 +97,12 @@ public final class MySQLPacketCodecEngine implements DatabasePacketCodecEngine<M
// CHECKSTYLE:ON
out.resetWriterIndex();
SQLException unknownSQLException = new UnknownSQLException(ex).toSQLException();
- new MySQLErrPacket(1, unknownSQLException.getErrorCode(), unknownSQLException.getSQLState(), unknownSQLException.getMessage()).write(payload);
+ new MySQLErrPacket(unknownSQLException.getErrorCode(), unknownSQLException.getSQLState(), unknownSQLException.getMessage()).write(payload);
} finally {
if (out.readableBytes() - PAYLOAD_LENGTH - SEQUENCE_LENGTH < MAX_PACKET_LENGTH) {
- updateMessageHeader(out, message.getSequenceId());
+ updateMessageHeader(out, context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get().getAndIncrement());
} else {
- // TODO Sequence ID of packets after this message may be duplicate with multi packets. Sequence ID should be generated by codec instead of provided by MySQLPacket.
- writeMultiPackets(context, out, message.getSequenceId());
+ writeMultiPackets(context, out);
}
}
}
@@ -113,14 +116,15 @@ public final class MySQLPacketCodecEngine implements DatabasePacketCodecEngine<M
byteBuf.setByte(3, sequenceId);
}
- private void writeMultiPackets(final ChannelHandlerContext context, final ByteBuf byteBuf, final int sequenceId) {
+ private void writeMultiPackets(final ChannelHandlerContext context, final ByteBuf byteBuf) {
int packetCount = (byteBuf.skipBytes(PAYLOAD_LENGTH + SEQUENCE_LENGTH).readableBytes() / MAX_PACKET_LENGTH) + 1;
CompositeByteBuf result = context.alloc().compositeBuffer(packetCount * 2);
+ AtomicInteger sequenceId = context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get();
for (int i = 0; i < packetCount; i++) {
ByteBuf header = context.alloc().ioBuffer(4, 4);
int packetLength = Math.min(byteBuf.readableBytes(), MAX_PACKET_LENGTH);
header.writeMediumLE(packetLength);
- header.writeByte(sequenceId + i);
+ header.writeByte(sequenceId.getAndIncrement());
result.addComponent(true, header);
if (packetLength > 0) {
result.addComponent(true, byteBuf.readRetainedSlice(packetLength));
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/constant/MySQLConstants.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/constant/MySQLConstants.java
index e8d363115b4..b23e35e8390 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/constant/MySQLConstants.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/constant/MySQLConstants.java
@@ -21,12 +21,16 @@ import io.netty.util.AttributeKey;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
+import java.util.concurrent.atomic.AtomicInteger;
+
/**
* MySQL constants.
*/
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public final class MySQLConstants {
+ public static final AttributeKey<AtomicInteger> MYSQL_SEQUENCE_ID = AttributeKey.valueOf("MYSQL_SEQUENCE_ID");
+
public static final AttributeKey<MySQLCharacterSet> MYSQL_CHARACTER_SET_ATTRIBUTE_KEY = AttributeKey.valueOf(MySQLCharacterSet.class.getName());
public static final AttributeKey<Integer> MYSQL_OPTION_MULTI_STATEMENTS = AttributeKey.valueOf("MYSQL_OPTION_MULTI_STATEMENTS");
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/MySQLPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/MySQLPacket.java
index 5e39965d0aa..23d59c217b9 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/MySQLPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/MySQLPacket.java
@@ -24,11 +24,4 @@ import org.apache.shardingsphere.db.protocol.packet.DatabasePacket;
* Database packet for MySQL.
*/
public interface MySQLPacket extends DatabasePacket<MySQLPacketPayload> {
-
- /**
- * Get sequence ID.
- *
- * @return sequence ID
- */
- int getSequenceId();
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/AbstractMySQLBinlogEventPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/AbstractMySQLBinlogEventPacket.java
index cde3274c0ae..70e784d98ac 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/AbstractMySQLBinlogEventPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/AbstractMySQLBinlogEventPacket.java
@@ -46,16 +46,11 @@ public abstract class AbstractMySQLBinlogEventPacket implements MySQLPacket, MyS
protected abstract void writeEvent(MySQLPacketPayload payload);
protected int getRemainBytesLength(final MySQLPacketPayload payload) {
- // minus checksum bytes, add seq id 1 byte, statusCode 1 byte(not include at event size)
- int alreadyReadIndex = binlogEventHeader.getEventSize() + 2 - binlogEventHeader.getChecksumLength();
+ // minus checksum bytes, add statusCode 1 byte(not include at event size)
+ int alreadyReadIndex = binlogEventHeader.getEventSize() + 1 - binlogEventHeader.getChecksumLength();
if (payload.getByteBuf().readerIndex() > alreadyReadIndex) {
return -1;
}
return alreadyReadIndex - payload.getByteBuf().readerIndex();
}
-
- @Override
- public final int getSequenceId() {
- return 0;
- }
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeader.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeader.java
index 2501070c9bc..78f48442069 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeader.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeader.java
@@ -74,9 +74,4 @@ public final class MySQLBinlogEventHeader implements MySQLPacket {
payload.writeInt4(logPos);
payload.writeInt2(flags);
}
-
- @Override
- public int getSequenceId() {
- return 0;
- }
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacket.java
index 878f6cea103..f34fdf82e96 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacket.java
@@ -38,9 +38,4 @@ public abstract class MySQLCommandPacket implements MySQLPacket, CommandPacket {
protected void doWrite(final MySQLPacketPayload payload) {
}
-
- @Override
- public final int getSequenceId() {
- return 0;
- }
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoader.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoader.java
index 47cdb3e63ea..1035c53da6e 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoader.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoader.java
@@ -17,7 +17,6 @@
package org.apache.shardingsphere.db.protocol.mysql.packet.command;
-import com.google.common.base.Preconditions;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@@ -35,7 +34,6 @@ public final class MySQLCommandPacketTypeLoader {
* @return command packet type for MySQL
*/
public static MySQLCommandPacketType getCommandPacketType(final MySQLPacketPayload payload) {
- Preconditions.checkArgument(0 == payload.readInt1(), "Sequence ID of MySQL command packet must be `0`");
return MySQLCommandPacketType.valueOf(payload.readInt1());
}
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41Packet.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41Packet.java
index 6cc743b66a7..1f306545242 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41Packet.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41Packet.java
@@ -18,7 +18,6 @@
package org.apache.shardingsphere.db.protocol.mysql.packet.command.query;
import com.google.common.base.Preconditions;
-import lombok.Getter;
import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLBinaryColumnType;
import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLServerInfo;
import org.apache.shardingsphere.db.protocol.mysql.packet.MySQLPacket;
@@ -39,9 +38,6 @@ public final class MySQLColumnDefinition41Packet implements MySQLPacket {
private static final int NEXT_LENGTH = 0x0c;
- @Getter
- private final int sequenceId;
-
private final int characterSet;
private final int flags;
@@ -64,8 +60,8 @@ public final class MySQLColumnDefinition41Packet implements MySQLPacket {
private final boolean containDefaultValues;
- public MySQLColumnDefinition41Packet(final int sequenceId, final ResultSetMetaData resultSetMetaData, final int columnIndex) throws SQLException {
- this(sequenceId, MySQLServerInfo.DEFAULT_CHARSET.getId(), resultSetMetaData.getSchemaName(columnIndex), resultSetMetaData.getTableName(columnIndex),
+ public MySQLColumnDefinition41Packet(final ResultSetMetaData resultSetMetaData, final int columnIndex) throws SQLException {
+ this(MySQLServerInfo.DEFAULT_CHARSET.getId(), resultSetMetaData.getSchemaName(columnIndex), resultSetMetaData.getTableName(columnIndex),
resultSetMetaData.getTableName(columnIndex), resultSetMetaData.getColumnLabel(columnIndex), resultSetMetaData.getColumnName(columnIndex),
resultSetMetaData.getColumnDisplaySize(columnIndex), MySQLBinaryColumnType.valueOfJDBCType(resultSetMetaData.getColumnType(columnIndex)), resultSetMetaData.getScale(columnIndex),
false);
@@ -76,16 +72,15 @@ public final class MySQLColumnDefinition41Packet implements MySQLPacket {
*
* @see <a href="https://github.com/apache/shardingsphere/issues/4358"></a>
*/
- public MySQLColumnDefinition41Packet(final int sequenceId, final int characterSet, final String schema, final String table, final String orgTable,
+ public MySQLColumnDefinition41Packet(final int characterSet, final String schema, final String table, final String orgTable,
final String name, final String orgName, final int columnLength, final MySQLBinaryColumnType columnType,
final int decimals, final boolean containDefaultValues) {
- this(sequenceId, characterSet, 0, schema, table, orgTable, name, orgName, columnLength, columnType, decimals, containDefaultValues);
+ this(characterSet, 0, schema, table, orgTable, name, orgName, columnLength, columnType, decimals, containDefaultValues);
}
- public MySQLColumnDefinition41Packet(final int sequenceId, final int characterSet, final int flags, final String schema, final String table, final String orgTable,
+ public MySQLColumnDefinition41Packet(final int characterSet, final int flags, final String schema, final String table, final String orgTable,
final String name, final String orgName, final int columnLength, final MySQLBinaryColumnType columnType,
final int decimals, final boolean containDefaultValues) {
- this.sequenceId = sequenceId;
this.characterSet = characterSet;
this.flags = flags;
this.schema = schema;
@@ -100,7 +95,6 @@ public final class MySQLColumnDefinition41Packet implements MySQLPacket {
}
public MySQLColumnDefinition41Packet(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(CATALOG.equals(payload.readStringLenenc()));
schema = payload.readStringLenenc();
table = payload.readStringLenenc();
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacket.java
index 64b206f0aef..c0df48cb4b4 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacket.java
@@ -31,12 +31,10 @@ import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@Getter
public final class MySQLFieldCountPacket implements MySQLPacket {
- private final int sequenceId;
-
private final int columnCount;
public MySQLFieldCountPacket(final MySQLPacketPayload payload) {
- this(payload.readInt1(), payload.readInt1());
+ this(payload.readInt1());
}
@Override
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacket.java
index f6871c3874b..e152647c6d6 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacket.java
@@ -17,7 +17,6 @@
package org.apache.shardingsphere.db.protocol.mysql.packet.command.query.binary.execute;
-import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.db.protocol.binary.BinaryCell;
import org.apache.shardingsphere.db.protocol.binary.BinaryRow;
@@ -37,9 +36,6 @@ public final class MySQLBinaryResultSetRowPacket implements MySQLPacket {
private static final int NULL_BITMAP_OFFSET = 2;
- @Getter
- private final int sequenceId;
-
private final BinaryRow row;
@Override
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareOKPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareOKPacket.java
index 0b81af5ea81..e82974df9a8 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareOKPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareOKPacket.java
@@ -17,7 +17,6 @@
package org.apache.shardingsphere.db.protocol.mysql.packet.command.query.binary.prepare;
-import lombok.Getter;
import lombok.RequiredArgsConstructor;
import org.apache.shardingsphere.db.protocol.mysql.packet.MySQLPacket;
import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@@ -32,9 +31,6 @@ public final class MySQLComStmtPrepareOKPacket implements MySQLPacket {
private static final int STATUS = 0x00;
- @Getter
- private final int sequenceId;
-
private final int statementId;
private final int columnCount;
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacket.java
index caf25f42208..f48b27de1ca 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacket.java
@@ -42,12 +42,9 @@ public final class MySQLTextResultSetRowPacket implements MySQLPacket {
private static final DateTimeFormatter DT_FMT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
- private final int sequenceId;
-
private final Collection<Object> data;
public MySQLTextResultSetRowPacket(final MySQLPacketPayload payload, final int columnCount) {
- sequenceId = payload.readInt1();
data = new ArrayList<>(columnCount);
for (int i = 0; i < columnCount; i++) {
data.add(payload.readStringLenenc());
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacket.java
index 8dbf2686966..618cb7b6e67 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacket.java
@@ -37,18 +37,15 @@ public final class MySQLEofPacket implements MySQLPacket {
*/
public static final int HEADER = 0xfe;
- private final int sequenceId;
-
private final int warnings;
private final int statusFlags;
- public MySQLEofPacket(final int sequenceId, final int statusFlags) {
- this(sequenceId, 0, statusFlags);
+ public MySQLEofPacket(final int statusFlags) {
+ this(0, statusFlags);
}
public MySQLEofPacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(HEADER == payload.readInt1(), "Header of MySQL EOF packet must be `0xfe`.");
warnings = payload.readInt2();
statusFlags = payload.readInt2();
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacket.java
index 09d25ddcaa0..144aee85f6a 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacket.java
@@ -40,20 +40,17 @@ public final class MySQLErrPacket implements MySQLPacket {
private static final String SQL_STATE_MARKER = "#";
- private final int sequenceId;
-
private final int errorCode;
private final String sqlState;
private final String errorMessage;
- public MySQLErrPacket(final int sequenceId, final VendorError vendorError, final Object... errorMessageArgs) {
- this(sequenceId, vendorError.getVendorCode(), vendorError.getSqlState().getValue(), String.format(vendorError.getReason(), errorMessageArgs));
+ public MySQLErrPacket(final VendorError vendorError, final Object... errorMessageArgs) {
+ this(vendorError.getVendorCode(), vendorError.getSqlState().getValue(), String.format(vendorError.getReason(), errorMessageArgs));
}
public MySQLErrPacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(HEADER == payload.readInt1(), "Header of MySQL ERR packet must be `0xff`.");
errorCode = payload.readInt2();
payload.readStringFix(1);
@@ -65,10 +62,8 @@ public final class MySQLErrPacket implements MySQLPacket {
public void write(final MySQLPacketPayload payload) {
payload.writeInt1(HEADER);
payload.writeInt2(errorCode);
- if (0 != sequenceId) {
- payload.writeStringFix(SQL_STATE_MARKER);
- payload.writeStringFix(sqlState);
- }
+ payload.writeStringFix(SQL_STATE_MARKER);
+ payload.writeStringFix(sqlState);
payload.writeStringEOF(errorMessage);
}
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacket.java
index cfadd643872..606b3311bbe 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacket.java
@@ -37,8 +37,6 @@ public final class MySQLOKPacket implements MySQLPacket {
*/
public static final int HEADER = 0x00;
- private final int sequenceId;
-
private final long affectedRows;
private final long lastInsertId;
@@ -49,16 +47,15 @@ public final class MySQLOKPacket implements MySQLPacket {
private final String info;
- public MySQLOKPacket(final int sequenceId, final int statusFlag) {
- this(sequenceId, 0L, 0L, statusFlag, 0, "");
+ public MySQLOKPacket(final int statusFlag) {
+ this(0L, 0L, statusFlag, 0, "");
}
- public MySQLOKPacket(final int sequenceId, final long affectedRows, final long lastInsertId, final int statusFlag) {
- this(sequenceId, affectedRows, lastInsertId, statusFlag, 0, "");
+ public MySQLOKPacket(final long affectedRows, final long lastInsertId, final int statusFlag) {
+ this(affectedRows, lastInsertId, statusFlag, 0, "");
}
public MySQLOKPacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(HEADER == payload.readInt1(), "Header of MySQL OK packet must be `0x00`.");
affectedRows = payload.readIntLenenc();
lastInsertId = payload.readIntLenenc();
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthMoreDataPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthMoreDataPacket.java
index 09c558334af..ad8d7d892f4 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthMoreDataPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthMoreDataPacket.java
@@ -38,13 +38,9 @@ public final class MySQLAuthMoreDataPacket implements MySQLPacket {
*/
public static final int HEADER = 0x01;
- @Getter
- private final int sequenceId;
-
private final byte[] pluginData;
public MySQLAuthMoreDataPacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(HEADER == payload.readInt1(), "Header of MySQL auth more data packet must be `0x01`.");
pluginData = payload.readStringEOFByBytes();
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacket.java
index d54496735b7..a4b04f0a9dd 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacket.java
@@ -38,9 +38,6 @@ public final class MySQLAuthSwitchRequestPacket implements MySQLPacket {
*/
public static final int HEADER = 0xfe;
- @Getter
- private final int sequenceId;
-
@Getter
private final String authPluginName;
@@ -48,7 +45,6 @@ public final class MySQLAuthSwitchRequestPacket implements MySQLPacket {
private final MySQLAuthPluginData authPluginData;
public MySQLAuthSwitchRequestPacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
Preconditions.checkArgument(HEADER == payload.readInt1(), "Header of MySQL auth switch request packet must be `0xfe`.");
authPluginName = payload.readStringNul();
String strAuthPluginData = payload.readStringNul();
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacket.java
index 7cb1ed6d6f2..73f19e0d171 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacket.java
@@ -31,13 +31,9 @@ import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@Getter
public final class MySQLAuthSwitchResponsePacket implements MySQLPacket {
- @Getter
- private final int sequenceId;
-
private final byte[] authPluginResponse;
public MySQLAuthSwitchResponsePacket(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
authPluginResponse = payload.readStringEOFByBytes();
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacket.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacket.java
index 07372b69bf5..40d21c5f7b8 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacket.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacket.java
@@ -64,7 +64,6 @@ public final class MySQLHandshakePacket implements MySQLPacket {
}
public MySQLHandshakePacket(final MySQLPacketPayload payload) {
- Preconditions.checkArgument(0 == payload.readInt1(), "Sequence ID of MySQL handshake packet must be `0`");
Preconditions.checkArgument(protocolVersion == payload.readInt1());
serverVersion = payload.readStringNul();
connectionId = payload.readInt4();
@@ -141,9 +140,4 @@ public final class MySQLHandshakePacket implements MySQLPacket {
private boolean isClientPluginAuth() {
return 0 != (capabilityFlagsUpper & MySQLCapabilityFlag.CLIENT_PLUGIN_AUTH.getValue() >> 16);
}
-
- @Override
- public int getSequenceId() {
- return 0;
- }
}
diff --git a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41Packet.java b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41Packet.java
index 5c3e55be953..74d1ebae1fb 100644
--- a/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41Packet.java
+++ b/db-protocol/mysql/src/main/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41Packet.java
@@ -35,8 +35,6 @@ import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
@Setter
public final class MySQLHandshakeResponse41Packet implements MySQLPacket {
- private final int sequenceId;
-
private final int maxPacketSize;
private final int characterSet;
@@ -52,7 +50,6 @@ public final class MySQLHandshakeResponse41Packet implements MySQLPacket {
private String authPluginName;
public MySQLHandshakeResponse41Packet(final MySQLPacketPayload payload) {
- sequenceId = payload.readInt1();
capabilityFlags = payload.readInt4();
maxPacketSize = payload.readInt4();
characterSet = payload.readInt1();
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngineTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngineTest.java
index 60b50286382..2b14aa5179b 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngineTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/codec/MySQLPacketCodecEngineTest.java
@@ -23,6 +23,7 @@ import io.netty.buffer.Unpooled;
import io.netty.buffer.UnpooledByteBufAllocator;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.AttributeKey;
+import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLConstants;
import org.apache.shardingsphere.db.protocol.mysql.packet.MySQLPacket;
import org.apache.shardingsphere.db.protocol.mysql.payload.MySQLPacketPayload;
import org.junit.Before;
@@ -38,6 +39,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
@@ -62,6 +64,7 @@ public final class MySQLPacketCodecEngineTest {
@Before
public void setup() {
when(context.channel().attr(AttributeKey.<Charset>valueOf(Charset.class.getName())).get()).thenReturn(StandardCharsets.UTF_8);
+ when(context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get()).thenReturn(new AtomicInteger());
}
@Test
@@ -79,7 +82,6 @@ public final class MySQLPacketCodecEngineTest {
when(byteBuf.markReaderIndex()).thenReturn(byteBuf);
when(byteBuf.readUnsignedMediumLE()).thenReturn(50);
when(byteBuf.readableBytes()).thenReturn(51);
- when(byteBuf.readRetainedSlice(51)).thenReturn(byteBuf);
List<Object> out = new LinkedList<>();
new MySQLPacketCodecEngine().decode(context, byteBuf, out);
assertThat(out.size(), is(1));
@@ -113,7 +115,7 @@ public final class MySQLPacketCodecEngineTest {
engine.decode(context, each, actual);
}
assertThat(actual.size(), is(1));
- assertThat(((ByteBuf) actual.get(0)).readableBytes(), is(1 << 24));
+ assertThat(((ByteBuf) actual.get(0)).readableBytes(), is((1 << 24) - 1));
}
private List<ByteBuf> preparePacketMoreThan16MB() {
@@ -130,7 +132,7 @@ public final class MySQLPacketCodecEngineTest {
when(byteBuf.markWriterIndex()).thenReturn(byteBuf);
when(byteBuf.readableBytes()).thenReturn(8);
MySQLPacket actualMessage = mock(MySQLPacket.class);
- when(actualMessage.getSequenceId()).thenReturn(1);
+ context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get().set(1);
new MySQLPacketCodecEngine().encode(context, actualMessage, byteBuf);
verify(byteBuf).writeInt(0);
verify(byteBuf).markWriterIndex();
@@ -170,7 +172,7 @@ public final class MySQLPacketCodecEngineTest {
RuntimeException ex = mock(RuntimeException.class);
MySQLPacket actualMessage = mock(MySQLPacket.class);
doThrow(ex).when(actualMessage).write(any(MySQLPacketPayload.class));
- when(actualMessage.getSequenceId()).thenReturn(2);
+ context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).get().set(2);
new MySQLPacketCodecEngine().encode(context, actualMessage, byteBuf);
verify(byteBuf).writeInt(0);
verify(byteBuf).markWriterIndex();
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeaderTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeaderTest.java
index 2aa80b9047b..e576e6a779c 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeaderTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/MySQLBinlogEventHeaderTest.java
@@ -42,7 +42,6 @@ public final class MySQLBinlogEventHeaderTest {
when(payload.readInt1()).thenReturn(MySQLBinlogEventType.UNKNOWN_EVENT.getValue());
when(payload.readInt2()).thenReturn(MySQLBinlogEventFlag.LOG_EVENT_BINLOG_IN_USE_F.getValue());
MySQLBinlogEventHeader actual = new MySQLBinlogEventHeader(payload, 4);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getTimestamp(), is(1234567890));
assertThat(actual.getEventType(), is(MySQLBinlogEventType.UNKNOWN_EVENT.getValue()));
assertThat(actual.getServerId(), is(123456));
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogFormatDescriptionEventPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogFormatDescriptionEventPacketTest.java
index e5525023cd1..988ce693b6f 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogFormatDescriptionEventPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogFormatDescriptionEventPacketTest.java
@@ -50,7 +50,6 @@ public final class MySQLBinlogFormatDescriptionEventPacketTest {
assertThat(actual.getMysqlServerVersion(), is(MYSQL_SERVER_VERSION.getBytes()));
assertThat(actual.getCreateTimestamp(), is(1234567890));
assertThat(actual.getEventHeaderLength(), is(19));
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getBinlogEventHeader(), is(binlogEventHeader));
}
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogRotateEventPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogRotateEventPacketTest.java
index ba4ee54c95a..42d63fcb0d9 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogRotateEventPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/management/MySQLBinlogRotateEventPacketTest.java
@@ -46,7 +46,6 @@ public final class MySQLBinlogRotateEventPacketTest {
when(payload.readStringFix(anyInt())).thenReturn("binlog-000001");
when(payload.getByteBuf()).thenReturn(Unpooled.buffer());
MySQLBinlogRotateEventPacket actual = new MySQLBinlogRotateEventPacket(binlogEventHeader, payload);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getBinlogEventHeader(), is(binlogEventHeader));
assertThat(actual.getPosition(), is(4L));
assertThat(actual.getNextBinlogName(), is("binlog-000001"));
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/row/MySQLBinlogTableMapEventPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/row/MySQLBinlogTableMapEventPacketTest.java
index 53573d7d0ee..238c5a7ba26 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/row/MySQLBinlogTableMapEventPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/binlog/row/MySQLBinlogTableMapEventPacketTest.java
@@ -61,6 +61,7 @@ public final class MySQLBinlogTableMapEventPacketTest {
when(payload.readIntLenenc()).thenReturn(4L);
when(payload.getByteBuf()).thenReturn(byteBuf);
when(byteBuf.readUnsignedShort()).thenReturn(10);
+ when(byteBuf.readerIndex()).thenReturn(1);
MySQLBinlogTableMapEventPacket actual = new MySQLBinlogTableMapEventPacket(binlogEventHeader, payload);
assertThat(actual.getTableId(), is(1L));
assertThat(actual.getFlags(), is(0));
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoaderTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoaderTest.java
index fef6f21e1fe..b8a5505fb0e 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoaderTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/MySQLCommandPacketTypeLoaderTest.java
@@ -30,14 +30,14 @@ public final class MySQLCommandPacketTypeLoaderTest {
@Test
public void assertGetCommandPacketType() {
MySQLPacketPayload payload = mock(MySQLPacketPayload.class);
- when(payload.readInt1()).thenReturn(0, MySQLCommandPacketType.COM_QUIT.getValue());
+ when(payload.readInt1()).thenReturn(MySQLCommandPacketType.COM_QUIT.getValue());
assertThat(MySQLCommandPacketTypeLoader.getCommandPacketType(payload), is(MySQLCommandPacketType.COM_QUIT));
}
@Test(expected = NullPointerException.class)
public void assertGetCommandPacketTypeError() {
MySQLPacketPayload payload = mock(MySQLPacketPayload.class);
- when(payload.readInt1()).thenReturn(0, 0x21);
+ when(payload.readInt1()).thenReturn(0x21);
MySQLCommandPacketTypeLoader.getCommandPacketType(payload);
}
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/admin/initdb/MySQLComInitDbPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/admin/initdb/MySQLComInitDbPacketTest.java
index ca91047ee53..33425e10c7e 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/admin/initdb/MySQLComInitDbPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/admin/initdb/MySQLComInitDbPacketTest.java
@@ -39,7 +39,6 @@ public final class MySQLComInitDbPacketTest {
public void assertNew() {
when(payload.readStringEOF()).thenReturn("logic_db");
MySQLComInitDbPacket actual = new MySQLComInitDbPacket(payload);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getSchema(), is("logic_db"));
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41PacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41PacketTest.java
index 40c1245f8ce..3f2e8e6580c 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41PacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLColumnDefinition41PacketTest.java
@@ -29,8 +29,6 @@ import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Types;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -52,21 +50,19 @@ public final class MySQLColumnDefinition41PacketTest {
when(resultSetMetaData.getColumnName(1)).thenReturn("id");
when(resultSetMetaData.getColumnDisplaySize(1)).thenReturn(10);
when(resultSetMetaData.getColumnType(1)).thenReturn(Types.INTEGER);
- MySQLColumnDefinition41Packet actual = new MySQLColumnDefinition41Packet(1, resultSetMetaData, 1);
- assertThat(actual.getSequenceId(), is(1));
+ MySQLColumnDefinition41Packet actual = new MySQLColumnDefinition41Packet(resultSetMetaData, 1);
actual.write(payload);
verifyWrite();
}
@Test
public void assertWriteWithPayload() {
- when(payload.readInt1()).thenReturn(1, MySQLBinaryColumnType.MYSQL_TYPE_LONG.getValue(), 0);
+ when(payload.readInt1()).thenReturn(MySQLBinaryColumnType.MYSQL_TYPE_LONG.getValue(), 0);
when(payload.readInt2()).thenReturn(MySQLServerInfo.DEFAULT_CHARSET.getId(), 0);
when(payload.readInt4()).thenReturn(10);
when(payload.readIntLenenc()).thenReturn(0x0cL);
when(payload.readStringLenenc()).thenReturn("def", "logic_db", "tbl", "tbl", "id", "id");
MySQLColumnDefinition41Packet actual = new MySQLColumnDefinition41Packet(payload);
- assertThat(actual.getSequenceId(), is(1));
actual.write(payload);
verifyWrite();
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacketTest.java
index 7a90c6a781d..c14428e4490 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/MySQLFieldCountPacketTest.java
@@ -36,9 +36,8 @@ public final class MySQLFieldCountPacketTest {
@Test
public void assertWrite() {
- when(payload.readInt1()).thenReturn(1, 3);
+ when(payload.readInt1()).thenReturn(3);
MySQLFieldCountPacket actual = new MySQLFieldCountPacket(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getColumnCount(), is(3));
actual.write(payload);
verify(payload).writeIntLenenc(3);
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacketTest.java
index 24408cd3fe1..c9e85fa66ac 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/execute/MySQLBinaryResultSetRowPacketTest.java
@@ -28,8 +28,6 @@ import org.mockito.junit.MockitoJUnitRunner;
import java.util.Arrays;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.verify;
@RunWith(MockitoJUnitRunner.class)
@@ -41,8 +39,7 @@ public final class MySQLBinaryResultSetRowPacketTest {
@Test
public void assertWrite() {
BinaryRow row = new BinaryRow(Arrays.asList(new BinaryCell(MySQLBinaryColumnType.MYSQL_TYPE_STRING, "value"), new BinaryCell(MySQLBinaryColumnType.MYSQL_TYPE_STRING, null)));
- MySQLBinaryResultSetRowPacket actual = new MySQLBinaryResultSetRowPacket(1, row);
- assertThat(actual.getSequenceId(), is(1));
+ MySQLBinaryResultSetRowPacket actual = new MySQLBinaryResultSetRowPacket(row);
actual.write(payload);
verify(payload).writeInt1(0x00);
verify(payload).writeInt1(0x08);
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareMySQLOKPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareMySQLOKPacketTest.java
index 751000115d4..7199b45e8b3 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareMySQLOKPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPrepareMySQLOKPacketTest.java
@@ -23,8 +23,6 @@ import org.junit.runner.RunWith;
import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -36,8 +34,7 @@ public final class MySQLComStmtPrepareMySQLOKPacketTest {
@Test
public void assertWrite() {
- MySQLComStmtPrepareOKPacket actual = new MySQLComStmtPrepareOKPacket(1, 1, 0, 1, 0);
- assertThat(actual.getSequenceId(), is(1));
+ MySQLComStmtPrepareOKPacket actual = new MySQLComStmtPrepareOKPacket(1, 0, 1, 0);
actual.write(payload);
verify(payload).writeInt1(0x00);
verify(payload, times(2)).writeInt2(0);
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPreparePacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPreparePacketTest.java
index 4c3d5b02b7a..3a888e28084 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPreparePacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/binary/prepare/MySQLComStmtPreparePacketTest.java
@@ -38,7 +38,6 @@ public final class MySQLComStmtPreparePacketTest {
public void assertNew() {
when(payload.readStringEOF()).thenReturn("SELECT id FROM tbl WHERE id=?");
MySQLComStmtPreparePacket actual = new MySQLComStmtPreparePacket(payload);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getSql(), is("SELECT id FROM tbl WHERE id=?"));
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacketTest.java
index 2efc7c48951..af01343832d 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/MySQLTextResultSetRowPacketTest.java
@@ -30,8 +30,6 @@ import java.time.format.DateTimeFormatter;
import java.util.Arrays;
import java.util.Collections;
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -44,10 +42,8 @@ public final class MySQLTextResultSetRowPacketTest {
@Test
public void assertNew() {
- when(payload.readInt1()).thenReturn(1);
when(payload.readStringLenenc()).thenReturn("value_a", null, "value_c");
MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(payload, 3);
- assertThat(actual.getSequenceId(), is(1));
verify(payload, times(3)).readStringLenenc();
}
@@ -55,7 +51,7 @@ public final class MySQLTextResultSetRowPacketTest {
public void assertWrite() {
long now = System.currentTimeMillis();
Timestamp timestamp = new Timestamp(now);
- MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(1, Arrays.asList(null, "value", BigDecimal.ONE, new byte[]{}, timestamp));
+ MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(Arrays.asList(null, "value", BigDecimal.ONE, new byte[]{}, timestamp));
actual.write(payload);
verify(payload).writeInt1(0xfb);
verify(payload).writeStringLenenc("value");
@@ -71,7 +67,7 @@ public final class MySQLTextResultSetRowPacketTest {
public void assertTimestampWithoutNanos() {
long now = System.currentTimeMillis() / 1000 * 1000;
Timestamp timestamp = new Timestamp(now);
- MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(1, Arrays.asList(null, "value", BigDecimal.ONE, new byte[]{}, timestamp));
+ MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(Arrays.asList(null, "value", BigDecimal.ONE, new byte[]{}, timestamp));
actual.write(payload);
verify(payload).writeInt1(0xfb);
verify(payload).writeStringLenenc("value");
@@ -83,7 +79,7 @@ public final class MySQLTextResultSetRowPacketTest {
public void assertLocalDateTime() {
String localDateTimeStr = "2021-08-23T17:30:30";
LocalDateTime time = LocalDateTime.parse(localDateTimeStr, DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"));
- MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(1, Collections.singletonList(time));
+ MySQLTextResultSetRowPacket actual = new MySQLTextResultSetRowPacket(Collections.singletonList(time));
actual.write(payload);
verify(payload).writeStringLenenc(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").format(LocalDateTime.parse(localDateTimeStr, DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"))));
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/query/MySQLComQueryPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/query/MySQLComQueryPacketTest.java
index 3a38a2e3243..f55b1186965 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/query/MySQLComQueryPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/command/query/text/query/MySQLComQueryPacketTest.java
@@ -39,7 +39,6 @@ public final class MySQLComQueryPacketTest {
public void assertNew() {
when(payload.readStringEOF()).thenReturn("SELECT id FROM tbl");
MySQLComQueryPacket actual = new MySQLComQueryPacket(payload, false);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getSql(), is("SELECT id FROM tbl"));
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacketTest.java
index e64d034a75b..5ba6905878d 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLEofPacketTest.java
@@ -36,26 +36,24 @@ public final class MySQLEofPacketTest {
private MySQLPacketPayload payload;
@Test
- public void assertNewEofPacketWithSequenceId() {
- MySQLEofPacket actual = new MySQLEofPacket(1, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
- assertThat(actual.getSequenceId(), is(1));
+ public void assertNewEofPacketWithStatusFlag() {
+ MySQLEofPacket actual = new MySQLEofPacket(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
assertThat(actual.getWarnings(), is(0));
assertThat(actual.getStatusFlags(), is(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
}
@Test
public void assertNewEofPacketWithPayload() {
- when(payload.readInt1()).thenReturn(5, MySQLEofPacket.HEADER);
+ when(payload.readInt1()).thenReturn(MySQLEofPacket.HEADER);
when(payload.readInt2()).thenReturn(0, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
MySQLEofPacket actual = new MySQLEofPacket(payload);
- assertThat(actual.getSequenceId(), is(5));
assertThat(actual.getWarnings(), is(0));
assertThat(actual.getStatusFlags(), is(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
}
@Test
public void assertWrite() {
- new MySQLEofPacket(1, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()).write(payload);
+ new MySQLEofPacket(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()).write(payload);
verify(payload).writeInt1(MySQLEofPacket.HEADER);
verify(payload).writeInt2(0);
verify(payload).writeInt2(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacketTest.java
index 85d01e1fb6b..7258c64c28a 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLErrPacketTest.java
@@ -37,8 +37,7 @@ public final class MySQLErrPacketTest {
@Test
public void assertNewErrPacketWithServerErrorCode() {
- MySQLErrPacket actual = new MySQLErrPacket(1, MySQLVendorError.ER_ACCESS_DENIED_ERROR, "root", "localhost", "root");
- assertThat(actual.getSequenceId(), is(1));
+ MySQLErrPacket actual = new MySQLErrPacket(MySQLVendorError.ER_ACCESS_DENIED_ERROR, "root", "localhost", "root");
assertThat(actual.getErrorCode(), is(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getVendorCode()));
assertThat(actual.getSqlState(), is(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getSqlState().getValue()));
assertThat(actual.getErrorMessage(), is(String.format(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getReason(), "root", "localhost", "root")));
@@ -46,13 +45,12 @@ public final class MySQLErrPacketTest {
@Test
public void assertNewErrPacketWithPayload() {
- when(payload.readInt1()).thenReturn(1, MySQLErrPacket.HEADER);
+ when(payload.readInt1()).thenReturn(MySQLErrPacket.HEADER);
when(payload.readInt2()).thenReturn(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getVendorCode());
when(payload.readStringFix(1)).thenReturn("#");
when(payload.readStringFix(5)).thenReturn(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getSqlState().getValue());
when(payload.readStringEOF()).thenReturn(String.format(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getReason(), "root", "localhost", "root"));
MySQLErrPacket actual = new MySQLErrPacket(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getVendorCode()));
assertThat(actual.getSqlState(), is(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getSqlState().getValue()));
assertThat(actual.getErrorMessage(), is(String.format(MySQLVendorError.ER_ACCESS_DENIED_ERROR.getReason(), "root", "localhost", "root")));
@@ -60,7 +58,7 @@ public final class MySQLErrPacketTest {
@Test
public void assertWrite() {
- new MySQLErrPacket(1, MySQLVendorError.ER_NO_DB_ERROR).write(payload);
+ new MySQLErrPacket(MySQLVendorError.ER_NO_DB_ERROR).write(payload);
verify(payload).writeInt1(MySQLErrPacket.HEADER);
verify(payload).writeInt2(1046);
verify(payload).writeStringFix("#");
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacketTest.java
index e292f094c5e..b634519952d 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/generic/MySQLOKPacketTest.java
@@ -36,9 +36,8 @@ public final class MySQLOKPacketTest {
private MySQLPacketPayload packetPayload;
@Test
- public void assertNewOKPacketWithSequenceId() {
- MySQLOKPacket actual = new MySQLOKPacket(1, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
- assertThat(actual.getSequenceId(), is(1));
+ public void assertNewOKPacketWithStatusFlag() {
+ MySQLOKPacket actual = new MySQLOKPacket(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
assertThat(actual.getAffectedRows(), is(0L));
assertThat(actual.getLastInsertId(), is(0L));
assertThat(actual.getStatusFlag(), is(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
@@ -48,8 +47,7 @@ public final class MySQLOKPacketTest {
@Test
public void assertNewOKPacketWithAffectedRowsAndLastInsertId() {
- MySQLOKPacket actual = new MySQLOKPacket(1, 100L, 9999L, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
- assertThat(actual.getSequenceId(), is(1));
+ MySQLOKPacket actual = new MySQLOKPacket(100L, 9999L, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue());
assertThat(actual.getAffectedRows(), is(100L));
assertThat(actual.getLastInsertId(), is(9999L));
assertThat(actual.getStatusFlag(), is(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
@@ -59,12 +57,11 @@ public final class MySQLOKPacketTest {
@Test
public void assertNewOKPacketWithPayload() {
- when(packetPayload.readInt1()).thenReturn(1, MySQLOKPacket.HEADER);
+ when(packetPayload.readInt1()).thenReturn(MySQLOKPacket.HEADER);
when(packetPayload.readInt2()).thenReturn(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue(), 0);
when(packetPayload.readIntLenenc()).thenReturn(100L, 9999L);
when(packetPayload.readStringEOF()).thenReturn("");
MySQLOKPacket actual = new MySQLOKPacket(packetPayload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getAffectedRows(), is(100L));
assertThat(actual.getLastInsertId(), is(9999L));
assertThat(actual.getStatusFlag(), is(MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()));
@@ -74,7 +71,7 @@ public final class MySQLOKPacketTest {
@Test
public void assertWrite() {
- new MySQLOKPacket(1, 100L, 9999L, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()).write(packetPayload);
+ new MySQLOKPacket(100L, 9999L, MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue()).write(packetPayload);
verify(packetPayload).writeInt1(MySQLOKPacket.HEADER);
verify(packetPayload).writeIntLenenc(100L);
verify(packetPayload).writeIntLenenc(9999L);
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacketTest.java
index ffbe13dab79..8724f3863eb 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchRequestPacketTest.java
@@ -40,7 +40,7 @@ public final class MySQLAuthSwitchRequestPacketTest {
@Test
public void assertWrite() {
when(authPluginData.getAuthenticationPluginData()).thenReturn(new byte[]{0x11, 0x22});
- MySQLAuthSwitchRequestPacket authSwitchRequestPacket = new MySQLAuthSwitchRequestPacket(1, "plugin", authPluginData);
+ MySQLAuthSwitchRequestPacket authSwitchRequestPacket = new MySQLAuthSwitchRequestPacket("plugin", authPluginData);
authSwitchRequestPacket.write(payload);
verify(payload).writeInt1(0xfe);
verify(payload, times(2)).writeStringNul(anyString());
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacketTest.java
index 8c19bef075b..d4344390d83 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLAuthSwitchResponsePacketTest.java
@@ -37,10 +37,8 @@ public final class MySQLAuthSwitchResponsePacketTest {
@Test
public void assertMySQLAuthSwitchResponsePacket() {
- when(payload.readInt1()).thenReturn(0x11);
when(payload.readStringEOFByBytes()).thenReturn(authPluginResponse);
MySQLAuthSwitchResponsePacket authSwitchResponsePacket = new MySQLAuthSwitchResponsePacket(payload);
- assertThat(authSwitchResponsePacket.getSequenceId(), is(0x11));
assertThat(authSwitchResponsePacket.getAuthPluginResponse(), is(authPluginResponse));
}
}
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacketTest.java
index 8cd64a6c459..a03a778889a 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakePacketTest.java
@@ -44,14 +44,13 @@ public final class MySQLHandshakePacketTest {
@Test
public void assertNewWithPayload() {
- when(payload.readInt1()).thenReturn(0, MySQLServerInfo.PROTOCOL_VERSION, MySQLServerInfo.DEFAULT_CHARSET.getId(), 0);
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.PROTOCOL_VERSION, MySQLServerInfo.DEFAULT_CHARSET.getId(), 0);
when(payload.readStringNul()).thenReturn(MySQLServerInfo.getDefaultServerVersion());
when(payload.readStringNulByBytes()).thenReturn(part1, part2);
when(payload.readInt4()).thenReturn(1000);
when(payload.readInt2()).thenReturn(
MySQLCapabilityFlag.calculateHandshakeCapabilityFlagsLower(), MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue(), MySQLCapabilityFlag.calculateHandshakeCapabilityFlagsUpper());
MySQLHandshakePacket actual = new MySQLHandshakePacket(payload);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getServerVersion(), is(MySQLServerInfo.getDefaultServerVersion()));
assertThat(actual.getCapabilityFlagsLower(), is(MySQLCapabilityFlag.calculateHandshakeCapabilityFlagsLower()));
assertThat(actual.getConnectionId(), is(1000));
@@ -65,14 +64,13 @@ public final class MySQLHandshakePacketTest {
@Test
public void assertNewWithClientPluginAuthPayload() {
- when(payload.readInt1()).thenReturn(0, MySQLServerInfo.PROTOCOL_VERSION, MySQLServerInfo.DEFAULT_CHARSET.getId(), 0);
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.PROTOCOL_VERSION, MySQLServerInfo.DEFAULT_CHARSET.getId(), 0);
when(payload.readStringNul()).thenReturn(MySQLServerInfo.getDefaultServerVersion(), MySQLAuthenticationMethod.SECURE_PASSWORD_AUTHENTICATION.getMethodName());
when(payload.readStringNulByBytes()).thenReturn(part1, part2);
when(payload.readInt4()).thenReturn(1000);
when(payload.readInt2()).thenReturn(
MySQLCapabilityFlag.calculateHandshakeCapabilityFlagsLower(), MySQLStatusFlag.SERVER_STATUS_AUTOCOMMIT.getValue(), MySQLCapabilityFlag.CLIENT_PLUGIN_AUTH.getValue() >> 16);
MySQLHandshakePacket actual = new MySQLHandshakePacket(payload);
- assertThat(actual.getSequenceId(), is(0));
assertThat(actual.getServerVersion(), is(MySQLServerInfo.getDefaultServerVersion()));
assertThat(actual.getCapabilityFlagsLower(), is(MySQLCapabilityFlag.calculateHandshakeCapabilityFlagsLower()));
assertThat(actual.getConnectionId(), is(1000));
diff --git a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41PacketTest.java b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41PacketTest.java
index ee70d0e787a..d26a627ec3c 100644
--- a/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41PacketTest.java
+++ b/db-protocol/mysql/src/test/java/org/apache/shardingsphere/db/protocol/mysql/packet/handshake/MySQLHandshakeResponse41PacketTest.java
@@ -27,8 +27,8 @@ import org.mockito.Mock;
import org.mockito.junit.MockitoJUnitRunner;
import static org.hamcrest.CoreMatchers.is;
-import static org.junit.Assert.assertNull;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertNull;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -40,12 +40,11 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertNewWithPayloadWithDatabase() {
- when(payload.readInt1()).thenReturn(1, MySQLServerInfo.DEFAULT_CHARSET.getId());
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.DEFAULT_CHARSET.getId());
when(payload.readInt4()).thenReturn(MySQLCapabilityFlag.CLIENT_CONNECT_WITH_DB.getValue(), 1000);
when(payload.readStringNul()).thenReturn("root", "sharding_db");
when(payload.readStringNulByBytes()).thenReturn(new byte[]{1});
MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getMaxPacketSize(), is(1000));
assertThat(actual.getCharacterSet(), is(MySQLServerInfo.DEFAULT_CHARSET.getId()));
assertThat(actual.getUsername(), is("root"));
@@ -58,12 +57,11 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertNewWithPayloadWithAuthPluginName() {
- when(payload.readInt1()).thenReturn(1, MySQLServerInfo.DEFAULT_CHARSET.getId());
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.DEFAULT_CHARSET.getId());
when(payload.readInt4()).thenReturn(MySQLCapabilityFlag.CLIENT_PLUGIN_AUTH.getValue(), 1000);
when(payload.readStringNul()).thenReturn("root", MySQLAuthenticationMethod.SECURE_PASSWORD_AUTHENTICATION.getMethodName());
when(payload.readStringNulByBytes()).thenReturn(new byte[]{1});
MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getMaxPacketSize(), is(1000));
assertThat(actual.getCharacterSet(), is(MySQLServerInfo.DEFAULT_CHARSET.getId()));
assertThat(actual.getUsername(), is("root"));
@@ -76,12 +74,11 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertNewWithPayloadWithClientPluginAuthLenencClientData() {
- when(payload.readInt1()).thenReturn(1, MySQLServerInfo.DEFAULT_CHARSET.getId());
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.DEFAULT_CHARSET.getId());
when(payload.readInt4()).thenReturn(MySQLCapabilityFlag.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA.getValue(), 1000);
when(payload.readStringNul()).thenReturn("root");
when(payload.readStringLenencByBytes()).thenReturn(new byte[]{1});
MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getMaxPacketSize(), is(1000));
assertThat(actual.getCharacterSet(), is(MySQLServerInfo.DEFAULT_CHARSET.getId()));
assertThat(actual.getUsername(), is("root"));
@@ -94,12 +91,11 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertNewWithPayloadWithClientSecureConnection() {
- when(payload.readInt1()).thenReturn(1, MySQLServerInfo.DEFAULT_CHARSET.getId(), 1);
+ when(payload.readInt1()).thenReturn(MySQLServerInfo.DEFAULT_CHARSET.getId(), 1);
when(payload.readInt4()).thenReturn(MySQLCapabilityFlag.CLIENT_SECURE_CONNECTION.getValue(), 1000);
when(payload.readStringNul()).thenReturn("root");
when(payload.readStringFixByBytes(1)).thenReturn(new byte[]{1});
MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(payload);
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getMaxPacketSize(), is(1000));
assertThat(actual.getCharacterSet(), is(MySQLServerInfo.DEFAULT_CHARSET.getId()));
assertThat(actual.getUsername(), is("root"));
@@ -112,7 +108,7 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertWriteWithDatabase() {
- MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(1, 100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
+ MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
actual.setAuthResponse(new byte[]{1});
actual.setDatabase("sharding_db");
actual.write(payload);
@@ -127,7 +123,7 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertWriteWithAuthPluginName() {
- MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(1, 100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
+ MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
actual.setAuthResponse(new byte[]{1});
actual.setAuthPluginName(MySQLAuthenticationMethod.SECURE_PASSWORD_AUTHENTICATION);
actual.write(payload);
@@ -142,7 +138,7 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertWriteWithClientPluginAuthLenencClientData() {
- MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(1, 100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
+ MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
actual.setCapabilityFlags(MySQLCapabilityFlag.CLIENT_PLUGIN_AUTH_LENENC_CLIENT_DATA.getValue());
actual.setAuthResponse(new byte[]{1});
actual.write(payload);
@@ -156,7 +152,7 @@ public final class MySQLHandshakeResponse41PacketTest {
@Test
public void assertWriteWithClientSecureConnection() {
- MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(1, 100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
+ MySQLHandshakeResponse41Packet actual = new MySQLHandshakeResponse41Packet(100, MySQLServerInfo.DEFAULT_CHARSET.getId(), "root");
actual.setCapabilityFlags(MySQLCapabilityFlag.CLIENT_SECURE_CONNECTION.getValue());
actual.setAuthResponse(new byte[]{1});
actual.write(payload);
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClient.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClient.java
index 25bdb8b066a..3aeffbee526 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClient.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClient.java
@@ -42,6 +42,7 @@ import org.apache.shardingsphere.data.pipeline.mysql.ingest.client.netty.MySQLNe
import org.apache.shardingsphere.data.pipeline.mysql.ingest.client.netty.MySQLNegotiatePackageDecoder;
import org.apache.shardingsphere.db.protocol.codec.PacketCodec;
import org.apache.shardingsphere.db.protocol.mysql.codec.MySQLPacketCodecEngine;
+import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLConstants;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.binlog.MySQLComBinlogDumpCommandPacket;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.binlog.MySQLComRegisterSlaveCommandPacket;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.query.text.query.MySQLComQueryPacket;
@@ -97,6 +98,7 @@ public final class MySQLClient {
@Override
protected void initChannel(final SocketChannel socketChannel) {
+ socketChannel.attr(MySQLConstants.MYSQL_SEQUENCE_ID).set(new AtomicInteger());
socketChannel.pipeline().addLast(new ChannelAttrInitializer());
socketChannel.pipeline().addLast(new PacketCodec(new MySQLPacketCodecEngine()));
socketChannel.pipeline().addLast(new MySQLNegotiatePackageDecoder());
@@ -119,6 +121,7 @@ public final class MySQLClient {
public synchronized boolean execute(final String queryString) {
responseCallback = new DefaultPromise<>(eventLoopGroup.next());
MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString, true);
+ resetSequenceID();
channel.writeAndFlush(comQueryPacket);
return null != waitExpectedResponse(MySQLOKPacket.class);
}
@@ -132,6 +135,7 @@ public final class MySQLClient {
public synchronized int executeUpdate(final String queryString) {
responseCallback = new DefaultPromise<>(eventLoopGroup.next());
MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString, false);
+ resetSequenceID();
channel.writeAndFlush(comQueryPacket);
return (int) Objects.requireNonNull(waitExpectedResponse(MySQLOKPacket.class)).getAffectedRows();
}
@@ -145,6 +149,7 @@ public final class MySQLClient {
public synchronized InternalResultSet executeQuery(final String queryString) {
responseCallback = new DefaultPromise<>(eventLoopGroup.next());
MySQLComQueryPacket comQueryPacket = new MySQLComQueryPacket(queryString, false);
+ resetSequenceID();
channel.writeAndFlush(comQueryPacket);
return waitExpectedResponse(InternalResultSet.class);
}
@@ -173,6 +178,7 @@ public final class MySQLClient {
InetSocketAddress localAddress = (InetSocketAddress) channel.localAddress();
MySQLComRegisterSlaveCommandPacket packet = new MySQLComRegisterSlaveCommandPacket(
connectInfo.getServerId(), localAddress.getHostName(), connectInfo.getUsername(), connectInfo.getPassword(), localAddress.getPort());
+ resetSequenceID();
channel.writeAndFlush(packet);
waitExpectedResponse(MySQLOKPacket.class);
}
@@ -201,6 +207,7 @@ public final class MySQLClient {
String tableKey = String.join(":", connectInfo.getHost(), String.valueOf(connectInfo.getPort()));
channel.pipeline().addLast(new MySQLBinlogEventPacketDecoder(checksumLength, GlobalTableMapEventMapping.getTableMapEventMap(tableKey)));
channel.pipeline().addLast(new MySQLBinlogEventHandler(getLastBinlogEvent(binlogFileName, binlogPosition)));
+ resetSequenceID();
channel.writeAndFlush(new MySQLComBinlogDumpCommandPacket((int) binlogPosition, connectInfo.getServerId(), binlogFileName));
}
@@ -211,6 +218,10 @@ public final class MySQLClient {
return result;
}
+ private void resetSequenceID() {
+ channel.attr(MySQLConstants.MYSQL_SEQUENCE_ID).get().set(0);
+ }
+
/**
* Poll binlog event.
*
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java
index 8da11701072..e30a2ac3b23 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoder.java
@@ -56,11 +56,10 @@ public final class MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder {
@Override
protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
- // readable bytes must greater + seqId(1b) + statusCode(1b) + header-length(19b) +
- while (in.readableBytes() >= 2 + MySQLBinlogEventHeader.MYSQL_BINLOG_EVENT_HEADER_LENGTH) {
+ // readable bytes must greater + statusCode(1b) + header-length(19b) +
+ while (in.readableBytes() >= 1 + MySQLBinlogEventHeader.MYSQL_BINLOG_EVENT_HEADER_LENGTH) {
in.markReaderIndex();
MySQLPacketPayload payload = new MySQLPacketPayload(in, ctx.channel().attr(CommonConstants.CHARSET_ATTRIBUTE_KEY).get());
- skipSequenceId(payload);
checkError(payload);
MySQLBinlogEventHeader binlogEventHeader = new MySQLBinlogEventHeader(payload, binlogContext.getChecksumLength());
// make sure event has complete body
@@ -106,7 +105,7 @@ public final class MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder {
return decodeDeleteRowsEventV2(binlogEventHeader, payload);
default:
PlaceholderEvent result = createPlaceholderEvent(binlogEventHeader);
- int remainDataLength = binlogEventHeader.getEventSize() + 2 - binlogEventHeader.getChecksumLength() - payload.getByteBuf().readerIndex();
+ int remainDataLength = binlogEventHeader.getEventSize() + 1 - binlogEventHeader.getChecksumLength() - payload.getByteBuf().readerIndex();
if (remainDataLength > 0) {
payload.skipReserved(remainDataLength);
}
@@ -114,10 +113,6 @@ public final class MySQLBinlogEventPacketDecoder extends ByteToMessageDecoder {
}
}
- private void skipSequenceId(final MySQLPacketPayload payload) {
- payload.readInt1();
- }
-
private void checkError(final MySQLPacketPayload payload) {
int statusCode = payload.readInt1();
if (255 == statusCode) {
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoder.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoder.java
index d9ca263dde1..b1bbab61b46 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoder.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoder.java
@@ -64,7 +64,7 @@ public final class MySQLCommandPacketDecoder extends ByteToMessageDecoder {
}
private void decodeFieldPacket(final MySQLPacketPayload payload) {
- if (MySQLEofPacket.HEADER != (payload.getByteBuf().getByte(1) & 0xff)) {
+ if (MySQLEofPacket.HEADER != (payload.getByteBuf().getByte(0) & 0xff)) {
internalResultSet.getFieldDescriptors().add(new MySQLColumnDefinition41Packet(payload));
} else {
new MySQLEofPacket(payload);
@@ -73,7 +73,7 @@ public final class MySQLCommandPacketDecoder extends ByteToMessageDecoder {
}
private void decodeRowDataPacket(final MySQLPacketPayload payload, final List<Object> out) {
- if (MySQLEofPacket.HEADER != (payload.getByteBuf().getByte(1) & 0xff)) {
+ if (MySQLEofPacket.HEADER != (payload.getByteBuf().getByte(0) & 0xff)) {
internalResultSet.getFieldValues().add(new MySQLTextResultSetRowPacket(payload, internalResultSet.getHeader().getColumnCount()));
} else {
new MySQLEofPacket(payload);
@@ -84,7 +84,7 @@ public final class MySQLCommandPacketDecoder extends ByteToMessageDecoder {
}
private void decodeResponsePacket(final MySQLPacketPayload payload, final List<Object> out) {
- switch (payload.getByteBuf().getByte(1) & 0xff) {
+ switch (payload.getByteBuf().getByte(0) & 0xff) {
case MySQLErrPacket.HEADER:
out.add(new MySQLErrPacket(payload));
break;
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandler.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandler.java
index 5261d2286c3..6805ab0828b 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandler.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandler.java
@@ -69,7 +69,7 @@ public final class MySQLNegotiateHandler extends ChannelInboundHandlerAdapter {
public void channelRead(final ChannelHandlerContext ctx, final Object msg) {
if (msg instanceof MySQLHandshakePacket) {
MySQLHandshakePacket handshake = (MySQLHandshakePacket) msg;
- MySQLHandshakeResponse41Packet handshakeResponsePacket = new MySQLHandshakeResponse41Packet(1, MAX_PACKET_SIZE, CHARACTER_SET, username);
+ MySQLHandshakeResponse41Packet handshakeResponsePacket = new MySQLHandshakeResponse41Packet(MAX_PACKET_SIZE, CHARACTER_SET, username);
handshakeResponsePacket.setAuthResponse(generateAuthResponse(handshake.getAuthPluginData().getAuthenticationPluginData()));
handshakeResponsePacket.setCapabilityFlags(generateClientCapability());
handshakeResponsePacket.setAuthPluginName(MySQLAuthenticationMethod.SECURE_PASSWORD_AUTHENTICATION);
@@ -92,7 +92,7 @@ public final class MySQLNegotiateHandler extends ChannelInboundHandlerAdapter {
default:
authPluginResponse = password.getBytes();
}
- ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(authSwitchRequest.getSequenceId() + 1, authPluginResponse));
+ ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(authPluginResponse));
seed = authSwitchRequest.getAuthPluginData().getAuthenticationPluginData();
return;
}
@@ -116,10 +116,10 @@ public final class MySQLNegotiateHandler extends ChannelInboundHandlerAdapter {
if (!publicKeyRequested) {
if (PERFORM_FULL_AUTHENTICATION == authMoreData.getPluginData()[0]) {
publicKeyRequested = true;
- ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(authMoreData.getSequenceId() + 1, new byte[]{REQUEST_PUBLIC_KEY}));
+ ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(new byte[]{REQUEST_PUBLIC_KEY}));
}
} else {
- ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(authMoreData.getSequenceId() + 1,
+ ctx.channel().writeAndFlush(new MySQLAuthSwitchResponsePacket(
PasswordEncryption.encryptWithRSAPublicKey(password, seed,
serverInfo.getServerVersion().greaterThanOrEqualTo(8, 0, 5) ? "RSA/ECB/OAEPWithSHA-1AndMGF1Padding" : "RSA/ECB/PKCS1Padding",
new String(authMoreData.getPluginData()))));
diff --git a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoder.java b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoder.java
index 18f6c2d70c4..1adcd261ad3 100644
--- a/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoder.java
+++ b/kernel/data-pipeline/dialect/mysql/src/main/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoder.java
@@ -59,7 +59,7 @@ public final class MySQLNegotiatePackageDecoder extends ByteToMessageDecoder {
}
private MySQLPacket decodeResponsePacket(final MySQLPacketPayload payload, final List<Object> out) {
- int header = payload.getByteBuf().getByte(1) & 0xff;
+ int header = payload.getByteBuf().getByte(0) & 0xff;
switch (header) {
case MySQLErrPacket.HEADER:
return new MySQLErrPacket(payload);
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClientTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClientTest.java
index 8b17111dd22..841f0c4413a 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClientTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/MySQLClientTest.java
@@ -21,8 +21,10 @@ import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.util.Attribute;
import io.netty.util.concurrent.Promise;
import org.apache.shardingsphere.data.pipeline.core.exception.job.BinlogSyncChannelAlreadyClosedException;
+import org.apache.shardingsphere.db.protocol.mysql.constant.MySQLConstants;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.binlog.MySQLComBinlogDumpCommandPacket;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.binlog.MySQLComRegisterSlaveCommandPacket;
import org.apache.shardingsphere.db.protocol.mysql.packet.command.query.text.query.MySQLComQueryPacket;
@@ -36,11 +38,13 @@ import org.mockito.internal.configuration.plugins.Plugins;
import org.mockito.junit.MockitoJUnitRunner;
import java.net.InetSocketAddress;
+import java.util.concurrent.atomic.AtomicInteger;
import static org.hamcrest.CoreMatchers.is;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -58,6 +62,7 @@ public final class MySQLClientTest {
private MySQLClient mysqlClient;
+ @SuppressWarnings("unchecked")
@Before
public void setUp() throws InterruptedException {
mysqlClient = new MySQLClient(new ConnectInfo(1, "host", 3306, "username", "password"));
@@ -69,6 +74,8 @@ public final class MySQLClientTest {
return null;
});
when(channel.localAddress()).thenReturn(new InetSocketAddress("host", 3306));
+ when(channel.attr(MySQLConstants.MYSQL_SEQUENCE_ID)).thenReturn(mock(Attribute.class));
+ when(channel.attr(MySQLConstants.MYSQL_SEQUENCE_ID).get()).thenReturn(new AtomicInteger());
}
@Test
@@ -82,7 +89,7 @@ public final class MySQLClientTest {
@Test
public void assertExecute() throws ReflectiveOperationException {
- mockChannelResponse(new MySQLOKPacket(0, 0));
+ mockChannelResponse(new MySQLOKPacket(0));
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("channel"), mysqlClient, channel);
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("eventLoopGroup"), mysqlClient, new NioEventLoopGroup(1));
assertTrue(mysqlClient.execute(""));
@@ -91,7 +98,7 @@ public final class MySQLClientTest {
@Test
public void assertExecuteUpdate() throws ReflectiveOperationException {
- MySQLOKPacket expected = new MySQLOKPacket(0, 10, 0, 0);
+ MySQLOKPacket expected = new MySQLOKPacket(10, 0, 0);
Plugins.getMemberAccessor().set(MySQLOKPacket.class.getDeclaredField("affectedRows"), expected, 10L);
mockChannelResponse(expected);
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("channel"), mysqlClient, channel);
@@ -117,7 +124,7 @@ public final class MySQLClientTest {
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("serverInfo"), mysqlClient, serverInfo);
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("channel"), mysqlClient, channel);
Plugins.getMemberAccessor().set(MySQLClient.class.getDeclaredField("eventLoopGroup"), mysqlClient, new NioEventLoopGroup(1));
- mockChannelResponse(new MySQLOKPacket(0, 0));
+ mockChannelResponse(new MySQLOKPacket(0));
mysqlClient.subscribe("", 4L);
verify(channel).writeAndFlush(ArgumentMatchers.any(MySQLComRegisterSlaveCommandPacket.class));
verify(channel).writeAndFlush(ArgumentMatchers.any(MySQLComBinlogDumpCommandPacket.class));
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoderTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoderTest.java
index 90e942189e0..c824c814ce3 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoderTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLBinlogEventPacketDecoderTest.java
@@ -86,7 +86,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
@Test
public void assertDecodeRotateEvent() {
ByteBuf byteBuf = Unpooled.buffer();
- byteBuf.writeBytes(StringUtil.decodeHexDump("01000000000004010000002c0000000000000020001a9100000000000062696e6c6f672e3030303032394af65c24"));
+ byteBuf.writeBytes(StringUtil.decodeHexDump("000000000004010000002c0000000000000020001a9100000000000062696e6c6f672e3030303032394af65c24"));
List<Object> decodedEvents = new LinkedList<>();
binlogEventPacketDecoder.decode(channelHandlerContext, byteBuf, decodedEvents);
assertThat(decodedEvents.size(), is(0));
@@ -96,7 +96,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
@Test
public void assertDecodeFormatDescriptionEvent() {
ByteBuf byteBuf = Unpooled.buffer();
- byteBuf.writeBytes(StringUtil.decodeHexDump("0200513aa8620f01000000790000000000000000000400382e302e323700000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+ byteBuf.writeBytes(StringUtil.decodeHexDump("00513aa8620f01000000790000000000000000000400382e302e323700000000000000000000000000000000000000000000000000000000000000000000000000000000000000"
+ "000000000013000d0008000000000400040000006100041a08000000080808020000000a0a0a2a2a001234000a280140081396"));
List<Object> decodedEvents = new LinkedList<>();
binlogEventPacketDecoder.decode(channelHandlerContext, byteBuf, decodedEvents);
@@ -108,7 +108,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
public void assertDecodeTableMapEvent() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
// the hex data is from binlog data, The first event used in Row Based Replication
- byteBuf.writeBytes(StringUtil.decodeHexDump("3400cb38a962130100000041000000be7d000000007b000000000001000464735f310009745f6f726465725f31000408030ff604c8000a020c0101000201e0ff0a9b3a"));
+ byteBuf.writeBytes(StringUtil.decodeHexDump("00cb38a962130100000041000000be7d000000007b000000000001000464735f310009745f6f726465725f31000408030ff604c8000a020c0101000201e0ff0a9b3a"));
binlogContext.getTableMap().put(123L, tableMapEventPacket);
List<Object> decodedEvents = new LinkedList<>();
binlogEventPacketDecoder.decode(channelHandlerContext, byteBuf, decodedEvents);
@@ -120,7 +120,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
public void assertDecodeWriteRowEvent() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
// the hex data is from INSERT INTO t_order(order_id, user_id, status, t_numeric) VALUES (1, 1, 'SUCCESS',null);
- byteBuf.writeBytes(StringUtil.decodeHexDump("30007a36a9621e0100000038000000bb7c000000007b00000000000100020004ff08010000000000000001000000075355434345535365eff9ff"));
+ byteBuf.writeBytes(StringUtil.decodeHexDump("007a36a9621e0100000038000000bb7c000000007b00000000000100020004ff08010000000000000001000000075355434345535365eff9ff"));
binlogContext.getTableMap().put(123L, tableMapEventPacket);
when(tableMapEventPacket.getColumnDefs()).thenReturn(columnDefs);
List<Object> decodedEvents = new LinkedList<>();
@@ -135,7 +135,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
public void assertDecodeUpdateRowEvent() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
// the hex data is from update t_order set status = 'updated' where order_id = 1;
- byteBuf.writeBytes(StringUtil.decodeHexDump("3500cb38a9621f010000004e0000000c7e000000007b00000000000100020004ffff08010000000000000001000000075355434345535308010000000000000001000000077570"
+ byteBuf.writeBytes(StringUtil.decodeHexDump("00cb38a9621f010000004e0000000c7e000000007b00000000000100020004ffff08010000000000000001000000075355434345535308010000000000000001000000077570"
+ "6461746564e78cee6c"));
binlogContext.getTableMap().put(123L, tableMapEventPacket);
when(tableMapEventPacket.getColumnDefs()).thenReturn(columnDefs);
@@ -152,7 +152,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
public void assertDecodeDeleteRowEvent() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
// delete from t_order where order_id = 1;
- byteBuf.writeBytes(StringUtil.decodeHexDump("51002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5"));
+ byteBuf.writeBytes(StringUtil.decodeHexDump("002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5"));
binlogContext.getTableMap().put(116L, tableMapEventPacket);
when(tableMapEventPacket.getColumnDefs()).thenReturn(columnDefs);
List<Object> decodedEvents = new LinkedList<>();
@@ -166,7 +166,7 @@ public final class MySQLBinlogEventPacketDecoderTest {
@Test
public void assertBinlogEventHeaderIncomplete() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
- byte[] completeData = StringUtil.decodeHexDump("51002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5");
+ byte[] completeData = StringUtil.decodeHexDump("002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5");
byteBuf.writeBytes(completeData);
// write incomplete event data
byteBuf.writeBytes(StringUtil.decodeHexDump("3400"));
@@ -181,9 +181,9 @@ public final class MySQLBinlogEventPacketDecoderTest {
@Test
public void assertBinlogEventBodyIncomplete() {
ByteBuf byteBuf = ByteBufAllocator.DEFAULT.buffer();
- byte[] completeData = StringUtil.decodeHexDump("51002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5");
+ byte[] completeData = StringUtil.decodeHexDump("002a80a862200100000038000000c569000000007400000000000100020004ff0801000000000000000100000007535543434553531c9580c5");
byteBuf.writeBytes(completeData);
- byte[] notCompleteData = StringUtil.decodeHexDump("3400cb38a962130100000041000000be7d000000007b000000000001000464735f310009745f6f726465725f31000408030f");
+ byte[] notCompleteData = StringUtil.decodeHexDump("00cb38a962130100000041000000be7d000000007b000000000001000464735f310009745f6f726465725f31000408030f");
byteBuf.writeBytes(notCompleteData);
List<Object> decodedEvents = new LinkedList<>();
binlogContext.getTableMap().put(116L, tableMapEventPacket);
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoderTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoderTest.java
index 23a383693f3..cdf063e8ceb 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoderTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLCommandPacketDecoderTest.java
@@ -64,7 +64,7 @@ public final class MySQLCommandPacketDecoderTest {
private ByteBuf mockOkPacket() {
when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) MySQLOKPacket.HEADER);
- when(byteBuf.getByte(1)).thenReturn((byte) MySQLOKPacket.HEADER);
+ when(byteBuf.getByte(0)).thenReturn((byte) MySQLOKPacket.HEADER);
return byteBuf;
}
@@ -77,8 +77,8 @@ public final class MySQLCommandPacketDecoderTest {
}
private ByteBuf mockErrPacket() {
- when(byteBuf.getByte(1)).thenReturn((byte) MySQLErrPacket.HEADER);
- when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) MySQLErrPacket.HEADER);
+ when(byteBuf.getByte(0)).thenReturn((byte) MySQLErrPacket.HEADER);
+ when(byteBuf.readUnsignedByte()).thenReturn((short) MySQLErrPacket.HEADER);
return byteBuf;
}
@@ -95,13 +95,13 @@ public final class MySQLCommandPacketDecoderTest {
}
private ByteBuf mockEmptyResultSetPacket() {
- when(byteBuf.getByte(1)).thenReturn((byte) 3);
+ when(byteBuf.getByte(0)).thenReturn((byte) 3);
return byteBuf;
}
private ByteBuf mockFieldDefinition41Packet() {
- when(byteBuf.getByte(1)).thenReturn((byte) 3);
- when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) 3, (short) 0x0c);
+ when(byteBuf.getByte(0)).thenReturn((byte) 3);
+ when(byteBuf.readUnsignedByte()).thenReturn((short) 3, (short) 0x0c);
when(byteBuf.readBytes(new byte[3])).then(invocationOnMock -> {
byte[] input = invocationOnMock.getArgument(0);
System.arraycopy("def".getBytes(), 0, input, 0, input.length);
@@ -111,8 +111,8 @@ public final class MySQLCommandPacketDecoderTest {
}
private ByteBuf mockEofPacket() {
- when(byteBuf.getByte(1)).thenReturn((byte) MySQLEofPacket.HEADER);
- when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) MySQLEofPacket.HEADER);
+ when(byteBuf.getByte(0)).thenReturn((byte) MySQLEofPacket.HEADER);
+ when(byteBuf.readUnsignedByte()).thenReturn((short) MySQLEofPacket.HEADER);
return byteBuf;
}
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandlerTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandlerTest.java
index 7e436bae8a8..661cc933f12 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandlerTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiateHandlerTest.java
@@ -86,7 +86,7 @@ public final class MySQLNegotiateHandlerTest {
@Test
public void assertChannelReadOkPacket() throws ReflectiveOperationException {
- MySQLOKPacket okPacket = new MySQLOKPacket(0, 0);
+ MySQLOKPacket okPacket = new MySQLOKPacket(0);
ServerInfo serverInfo = new ServerInfo();
Plugins.getMemberAccessor().set(MySQLNegotiateHandler.class.getDeclaredField("serverInfo"), mysqlNegotiateHandler, serverInfo);
mysqlNegotiateHandler.channelRead(channelHandlerContext, okPacket);
@@ -96,7 +96,7 @@ public final class MySQLNegotiateHandlerTest {
@Test(expected = RuntimeException.class)
public void assertChannelReadErrorPacket() {
- MySQLErrPacket errorPacket = new MySQLErrPacket(0, MySQLVendorError.ER_NO_DB_ERROR);
+ MySQLErrPacket errorPacket = new MySQLErrPacket(MySQLVendorError.ER_NO_DB_ERROR);
mysqlNegotiateHandler.channelRead(channelHandlerContext, errorPacket);
}
}
diff --git a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoderTest.java b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoderTest.java
index 606f1ca70b7..086b037ae43 100644
--- a/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoderTest.java
+++ b/kernel/data-pipeline/dialect/mysql/src/test/java/org/apache/shardingsphere/data/pipeline/mysql/ingest/client/netty/MySQLNegotiatePackageDecoderTest.java
@@ -71,7 +71,7 @@ public final class MySQLNegotiatePackageDecoderTest {
}
private ByteBuf mockHandshakePacket() {
- String handshakePacket = "000a352e372e32312d6c6f6700090000004a592a1f725a0d0900fff7210200ff8115000000000000000000001a437b30323a4d2b514b5870006d"
+ String handshakePacket = "0a352e372e32312d6c6f6700090000004a592a1f725a0d0900fff7210200ff8115000000000000000000001a437b30323a4d2b514b5870006d"
+ "7973716c5f6e61746976655f70617373776f72640000000002000000";
byte[] handshakePacketBytes = ByteBufUtil.decodeHexDump(handshakePacket);
ByteBuf result = Unpooled.buffer(handshakePacketBytes.length);
@@ -103,8 +103,8 @@ public final class MySQLNegotiatePackageDecoderTest {
}
private ByteBuf authSwitchRequestPacket() {
- when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) MySQLAuthSwitchRequestPacket.HEADER);
- when(byteBuf.getByte(1)).thenReturn((byte) MySQLAuthSwitchRequestPacket.HEADER);
+ when(byteBuf.readUnsignedByte()).thenReturn((short) MySQLAuthSwitchRequestPacket.HEADER);
+ when(byteBuf.getByte(0)).thenReturn((byte) MySQLAuthSwitchRequestPacket.HEADER);
when(byteBuf.bytesBefore((byte) 0)).thenReturn(20);
return byteBuf;
}
@@ -119,8 +119,8 @@ public final class MySQLNegotiatePackageDecoderTest {
}
private ByteBuf authMoreDataPacket() {
- when(byteBuf.readUnsignedByte()).thenReturn((short) 0, (short) MySQLAuthMoreDataPacket.HEADER);
- when(byteBuf.getByte(1)).thenReturn((byte) MySQLAuthMoreDataPacket.HEADER);
+ when(byteBuf.readUnsignedByte()).thenReturn((short) MySQLAuthMoreDataPacket.HEADER);
+ when(byteBuf.getByte(0)).thenReturn((byte) MySQLAuthMoreDataPacket.HEADER);
return byteBuf;
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngine.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngine.java
index c28d5ce332c..4d5b4a4c241 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngine.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngine.java
@@ -45,6 +45,7 @@ import org.apache.shardingsphere.proxy.frontend.mysql.authentication.authenticat
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* Authentication engine for MySQL.
@@ -57,14 +58,13 @@ public final class MySQLAuthenticationEngine implements AuthenticationEngine {
private MySQLConnectionPhase connectionPhase = MySQLConnectionPhase.INITIAL_HANDSHAKE;
- private int sequenceId;
-
private byte[] authResponse;
private AuthenticationResult currentAuthResult;
@Override
public int handshake(final ChannelHandlerContext context) {
+ context.channel().attr(MySQLConstants.MYSQL_SEQUENCE_ID).set(new AtomicInteger());
int result = ConnectionIdGenerator.getInstance().nextId();
connectionPhase = MySQLConnectionPhase.AUTH_PHASE_FAST_PATH;
context.writeAndFlush(new MySQLHandshakePacket(result, authenticationHandler.getAuthPluginData()));
@@ -88,26 +88,25 @@ public final class MySQLAuthenticationEngine implements AuthenticationEngine {
context.close();
return AuthenticationResultBuilder.continued();
}
- context.writeAndFlush(new MySQLOKPacket(++sequenceId, DEFAULT_STATUS_FLAG));
+ context.writeAndFlush(new MySQLOKPacket(DEFAULT_STATUS_FLAG));
return AuthenticationResultBuilder.finished(currentAuthResult.getUsername(), getHostAddress(context), currentAuthResult.getDatabase());
}
private AuthenticationResult authPhaseFastPath(final ChannelHandlerContext context, final PacketPayload payload) {
MySQLHandshakeResponse41Packet packet = new MySQLHandshakeResponse41Packet((MySQLPacketPayload) payload);
authResponse = packet.getAuthResponse();
- sequenceId = packet.getSequenceId();
MySQLCharacterSet characterSet = MySQLCharacterSet.findById(packet.getCharacterSet());
context.channel().attr(CommonConstants.CHARSET_ATTRIBUTE_KEY).set(characterSet.getCharset());
context.channel().attr(MySQLConstants.MYSQL_CHARACTER_SET_ATTRIBUTE_KEY).set(characterSet);
if (!Strings.isNullOrEmpty(packet.getDatabase()) && !ProxyContext.getInstance().databaseExists(packet.getDatabase())) {
- context.writeAndFlush(new MySQLErrPacket(++sequenceId, MySQLVendorError.ER_BAD_DB_ERROR, packet.getDatabase()));
+ context.writeAndFlush(new MySQLErrPacket(MySQLVendorError.ER_BAD_DB_ERROR, packet.getDatabase()));
context.close();
return AuthenticationResultBuilder.continued();
}
MySQLAuthenticator authenticator = authenticationHandler.getAuthenticator(packet.getUsername(), getHostAddress(context));
if (isClientPluginAuth(packet) && !authenticator.getAuthenticationMethodName().equals(packet.getAuthPluginName())) {
connectionPhase = MySQLConnectionPhase.AUTHENTICATION_METHOD_MISMATCH;
- context.writeAndFlush(new MySQLAuthSwitchRequestPacket(++sequenceId, authenticator.getAuthenticationMethodName(), authenticationHandler.getAuthPluginData()));
+ context.writeAndFlush(new MySQLAuthSwitchRequestPacket(authenticator.getAuthenticationMethodName(), authenticationHandler.getAuthPluginData()));
return AuthenticationResultBuilder.continued(packet.getUsername(), getHostAddress(context), packet.getDatabase());
}
return AuthenticationResultBuilder.finished(packet.getUsername(), getHostAddress(context), packet.getDatabase());
@@ -119,14 +118,13 @@ public final class MySQLAuthenticationEngine implements AuthenticationEngine {
private void authenticationMethodMismatch(final MySQLPacketPayload payload) {
MySQLAuthSwitchResponsePacket packet = new MySQLAuthSwitchResponsePacket(payload);
- sequenceId = packet.getSequenceId();
authResponse = packet.getAuthPluginResponse();
}
private MySQLErrPacket createErrorPacket(final MySQLVendorError vendorError, final ChannelHandlerContext context) {
return MySQLVendorError.ER_DBACCESS_DENIED_ERROR == vendorError
- ? new MySQLErrPacket(++sequenceId, MySQLVendorError.ER_DBACCESS_DENIED_ERROR, currentAuthResult.getUsername(), getHostAddress(context), currentAuthResult.getDatabase())
- : new MySQLErrPacket(++sequenceId, MySQLVendorError.ER_ACCESS_DENIED_ERROR, currentAuthResult.getUsername(), getHostAddress(context), getErrorMessage());
+ ? new MySQLErrPacket(MySQLVendorError.ER_DBACCESS_DENIED_ERROR, currentAuthResult.getUsername(), getHostAddress(context), currentAuthResult.getDatabase())
+ : new MySQLErrPacket(MySQLVendorError.ER_ACCESS_DENIED_ERROR, currentAuthResult.getUsername(), getHostAddress(context), getErrorMessage());
}
private String getErrorMessage() {
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/MySQLCommandExecuteEngine.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/MySQLCommandExecuteEngine.java
index 307d9a81c30..c00f514e7b9 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/MySQLCommandExecuteEngine.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/MySQLCommandExecuteEngine.java
@@ -82,7 +82,6 @@ public final class MySQLCommandExecuteEngine implements CommandExecuteEngine {
}
int count = 0;
int flushThreshold = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getProps().<Integer>getValue(ConfigurationPropertyKey.PROXY_FRONTEND_FLUSH_THRESHOLD);
- int currentSequenceId = 0;
while (queryCommandExecutor.next()) {
count++;
while (!context.channel().isWritable() && context.channel().isActive()) {
@@ -95,8 +94,7 @@ public final class MySQLCommandExecuteEngine implements CommandExecuteEngine {
context.flush();
count = 0;
}
- currentSequenceId++;
}
- context.write(new MySQLEofPacket(++currentSequenceId + headerPackagesCount, ServerStatusFlagCalculator.calculateFor(backendConnection.getConnectionSession())));
+ context.write(new MySQLEofPacket(ServerStatusFlagCalculator.calculateFor(backendConnection.getConnectionSession())));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComResetConnectionExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComResetConnectionExecutor.java
index d068432ec2c..c62c770936d 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComResetConnectionExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComResetConnectionExecutor.java
@@ -44,6 +44,6 @@ public final class MySQLComResetConnectionExecutor implements CommandExecutor {
connectionSession.setDefaultIsolationLevel(null);
connectionSession.setIsolationLevel(null);
connectionSession.getServerPreparedStatementRegistry().clear();
- return Collections.singletonList(new MySQLOKPacket(1, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ return Collections.singletonList(new MySQLOKPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComSetOptionExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComSetOptionExecutor.java
index 5476eb433a1..d5bef2707c8 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComSetOptionExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/MySQLComSetOptionExecutor.java
@@ -42,6 +42,6 @@ public final class MySQLComSetOptionExecutor implements CommandExecutor {
@Override
public Collection<DatabasePacket<?>> execute() {
connectionSession.getAttributeMap().attr(MySQLConstants.MYSQL_OPTION_MULTI_STATEMENTS).set(packet.getValue());
- return Collections.singletonList(new MySQLOKPacket(1, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ return Collections.singletonList(new MySQLOKPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/initdb/MySQLComInitDbExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/initdb/MySQLComInitDbExecutor.java
index 65450a1014f..91415292e13 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/initdb/MySQLComInitDbExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/initdb/MySQLComInitDbExecutor.java
@@ -49,7 +49,7 @@ public final class MySQLComInitDbExecutor implements CommandExecutor {
String databaseName = SQLUtil.getExactlyValue(packet.getSchema());
if (ProxyContext.getInstance().databaseExists(databaseName) && SQLCheckEngine.check(databaseName, getRules(databaseName), connectionSession.getGrantee())) {
connectionSession.setCurrentDatabase(packet.getSchema());
- return Collections.singletonList(new MySQLOKPacket(1, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ return Collections.singletonList(new MySQLOKPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
}
throw new UnknownDatabaseException(packet.getSchema());
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/ping/MySQLComPingExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/ping/MySQLComPingExecutor.java
index 94733519681..eac7912a147 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/ping/MySQLComPingExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/ping/MySQLComPingExecutor.java
@@ -37,6 +37,6 @@ public final class MySQLComPingExecutor implements CommandExecutor {
@Override
public Collection<DatabasePacket<?>> execute() {
- return Collections.singletonList(new MySQLOKPacket(1, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ return Collections.singletonList(new MySQLOKPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/quit/MySQLComQuitExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/quit/MySQLComQuitExecutor.java
index 838c84fb0ec..004605d83a4 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/quit/MySQLComQuitExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/admin/quit/MySQLComQuitExecutor.java
@@ -31,6 +31,6 @@ public final class MySQLComQuitExecutor implements CommandExecutor {
@Override
public Collection<DatabasePacket<?>> execute() {
- return Collections.singletonList(new MySQLOKPacket(1, 0));
+ return Collections.singletonList(new MySQLOKPacket(0));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutor.java
index bd8737001da..3bf5b68b62d 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutor.java
@@ -66,8 +66,6 @@ public final class MySQLComStmtExecuteExecutor implements QueryCommandExecutor {
@Getter
private ResponseType responseType;
- private int currentSequenceId;
-
@Override
public Collection<DatabasePacket<?>> execute() throws SQLException {
MySQLServerPreparedStatement preparedStatement = updateAndGetPreparedStatement();
@@ -96,9 +94,7 @@ public final class MySQLComStmtExecuteExecutor implements QueryCommandExecutor {
private Collection<DatabasePacket<?>> processQuery(final QueryResponseHeader queryResponseHeader) {
responseType = ResponseType.QUERY;
int characterSet = connectionSession.getAttributeMap().attr(MySQLConstants.MYSQL_CHARACTER_SET_ATTRIBUTE_KEY).get().getId();
- Collection<DatabasePacket<?>> result = ResponsePacketBuilder.buildQueryResponsePackets(queryResponseHeader, characterSet, ServerStatusFlagCalculator.calculateFor(connectionSession));
- currentSequenceId = result.size();
- return result;
+ return ResponsePacketBuilder.buildQueryResponsePackets(queryResponseHeader, characterSet, ServerStatusFlagCalculator.calculateFor(connectionSession));
}
private Collection<DatabasePacket<?>> processUpdate(final UpdateResponseHeader updateResponseHeader) {
@@ -114,7 +110,7 @@ public final class MySQLComStmtExecuteExecutor implements QueryCommandExecutor {
@Override
public MySQLPacket getQueryRowPacket() throws SQLException {
QueryResponseRow queryResponseRow = proxyBackendHandler.getRowData();
- return new MySQLBinaryResultSetRowPacket(++currentSequenceId, createBinaryRow(queryResponseRow));
+ return new MySQLBinaryResultSetRowPacket(createBinaryRow(queryResponseRow));
}
private BinaryRow createBinaryRow(final QueryResponseRow queryResponseRow) {
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLComStmtPrepareExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLComStmtPrepareExecutor.java
index a2f6362ceff..cf0bc9013e4 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLComStmtPrepareExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/prepare/MySQLComStmtPrepareExecutor.java
@@ -55,8 +55,6 @@ public final class MySQLComStmtPrepareExecutor implements CommandExecutor {
private final ConnectionSession connectionSession;
- private int currentSequenceId;
-
@Override
public Collection<DatabasePacket<?>> execute() {
failedIfContainsMultiStatements();
@@ -94,7 +92,7 @@ public final class MySQLComStmtPrepareExecutor implements CommandExecutor {
private Collection<DatabasePacket<?>> createPackets(final int statementId, final int projectionCount, final int parameterCount) {
Collection<DatabasePacket<?>> result = new LinkedList<>();
- result.add(new MySQLComStmtPrepareOKPacket(++currentSequenceId, statementId, projectionCount, parameterCount, 0));
+ result.add(new MySQLComStmtPrepareOKPacket(statementId, projectionCount, parameterCount, 0));
int characterSet = connectionSession.getAttributeMap().attr(MySQLConstants.MYSQL_CHARACTER_SET_ATTRIBUTE_KEY).get().getId();
int statusFlags = ServerStatusFlagCalculator.calculateFor(connectionSession);
if (parameterCount > 0) {
@@ -109,18 +107,18 @@ public final class MySQLComStmtPrepareExecutor implements CommandExecutor {
private Collection<DatabasePacket<?>> createParameterColumnDefinition41Packets(final int parameterCount, final int characterSet, final int statusFlags) {
Collection<DatabasePacket<?>> result = new LinkedList<>();
for (int i = 0; i < parameterCount; i++) {
- result.add(new MySQLColumnDefinition41Packet(++currentSequenceId, characterSet, "", "", "", "?", "", 0, MySQLBinaryColumnType.MYSQL_TYPE_VAR_STRING, 0, false));
+ result.add(new MySQLColumnDefinition41Packet(characterSet, "", "", "", "?", "", 0, MySQLBinaryColumnType.MYSQL_TYPE_VAR_STRING, 0, false));
}
- result.add(new MySQLEofPacket(++currentSequenceId, statusFlags));
+ result.add(new MySQLEofPacket(statusFlags));
return result;
}
private Collection<DatabasePacket<?>> createProjectionColumnDefinition41Packets(final int projectionCount, final int characterSet, final int statusFlags) {
Collection<DatabasePacket<?>> result = new LinkedList<>();
for (int i = 0; i < projectionCount; i++) {
- result.add(new MySQLColumnDefinition41Packet(++currentSequenceId, characterSet, "", "", "", "", "", 0, MySQLBinaryColumnType.MYSQL_TYPE_VAR_STRING, 0, false));
+ result.add(new MySQLColumnDefinition41Packet(characterSet, "", "", "", "", "", 0, MySQLBinaryColumnType.MYSQL_TYPE_VAR_STRING, 0, false));
}
- result.add(new MySQLEofPacket(++currentSequenceId, statusFlags));
+ result.add(new MySQLEofPacket(statusFlags));
return result;
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/reset/MySQLComStmtResetExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/reset/MySQLComStmtResetExecutor.java
index d4d0fe9bd58..6635b0c700e 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/reset/MySQLComStmtResetExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/reset/MySQLComStmtResetExecutor.java
@@ -42,6 +42,6 @@ public final class MySQLComStmtResetExecutor implements CommandExecutor {
@Override
public Collection<DatabasePacket<?>> execute() {
connectionSession.getServerPreparedStatementRegistry().<MySQLServerPreparedStatement>getPreparedStatement(packet.getStatementId()).getLongData().clear();
- return Collections.singleton(new MySQLOKPacket(1, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ return Collections.singleton(new MySQLOKPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/builder/ResponsePacketBuilder.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/builder/ResponsePacketBuilder.java
index 6acf6ffa500..e5134010a8a 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/builder/ResponsePacketBuilder.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/builder/ResponsePacketBuilder.java
@@ -55,14 +55,13 @@ public final class ResponsePacketBuilder {
*/
public static Collection<DatabasePacket<?>> buildQueryResponsePackets(final QueryResponseHeader queryResponseHeader, final int characterSet, final int statusFlags) {
Collection<DatabasePacket<?>> result = new LinkedList<>();
- int sequenceId = 0;
List<QueryHeader> queryHeaders = queryResponseHeader.getQueryHeaders();
- result.add(new MySQLFieldCountPacket(++sequenceId, queryHeaders.size()));
+ result.add(new MySQLFieldCountPacket(queryHeaders.size()));
for (QueryHeader each : queryHeaders) {
- result.add(new MySQLColumnDefinition41Packet(++sequenceId, characterSet, getColumnFieldDetailFlag(each), each.getSchema(), each.getTable(), each.getTable(),
+ result.add(new MySQLColumnDefinition41Packet(characterSet, getColumnFieldDetailFlag(each), each.getSchema(), each.getTable(), each.getTable(),
each.getColumnLabel(), each.getColumnName(), each.getColumnLength(), MySQLBinaryColumnType.valueOfJDBCType(each.getColumnType()), each.getDecimals(), false));
}
- result.add(new MySQLEofPacket(++sequenceId, statusFlags));
+ result.add(new MySQLEofPacket(statusFlags));
return result;
}
@@ -94,6 +93,6 @@ public final class ResponsePacketBuilder {
* @return update response packets
*/
public static Collection<DatabasePacket<?>> buildUpdateResponsePackets(final UpdateResponseHeader updateResponseHeader, final int serverStatusFlag) {
- return Collections.singletonList(new MySQLOKPacket(1, updateResponseHeader.getUpdateCount(), updateResponseHeader.getLastInsertId(), serverStatusFlag));
+ return Collections.singletonList(new MySQLOKPacket(updateResponseHeader.getUpdateCount(), updateResponseHeader.getLastInsertId(), serverStatusFlag));
}
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/fieldlist/MySQLComFieldListPacketExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/fieldlist/MySQLComFieldListPacketExecutor.java
index 012522aa928..1be3fde6d74 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/fieldlist/MySQLComFieldListPacketExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/fieldlist/MySQLComFieldListPacketExecutor.java
@@ -58,8 +58,6 @@ public final class MySQLComFieldListPacketExecutor implements CommandExecutor {
private DatabaseCommunicationEngine databaseCommunicationEngine;
- private int currentSequenceId;
-
@Override
public Collection<DatabasePacket<?>> execute() throws SQLException {
String databaseName = connectionSession.getDefaultDatabaseName();
@@ -81,9 +79,9 @@ public final class MySQLComFieldListPacketExecutor implements CommandExecutor {
while (databaseCommunicationEngine.next()) {
String columnName = databaseCommunicationEngine.getRowData().getCells().iterator().next().getData().toString();
result.add(new MySQLColumnDefinition41Packet(
- ++currentSequenceId, characterSet, databaseName, packet.getTable(), packet.getTable(), columnName, columnName, 100, MySQLBinaryColumnType.MYSQL_TYPE_VARCHAR, 0, true));
+ characterSet, databaseName, packet.getTable(), packet.getTable(), columnName, columnName, 100, MySQLBinaryColumnType.MYSQL_TYPE_VARCHAR, 0, true));
}
- result.add(new MySQLEofPacket(++currentSequenceId, ServerStatusFlagCalculator.calculateFor(connectionSession)));
+ result.add(new MySQLEofPacket(ServerStatusFlagCalculator.calculateFor(connectionSession)));
return result;
}
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/query/MySQLComQueryPacketExecutor.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/query/MySQLComQueryPacketExecutor.java
index b07917a09d5..a157d33dd24 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/query/MySQLComQueryPacketExecutor.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/text/query/MySQLComQueryPacketExecutor.java
@@ -62,8 +62,6 @@ public final class MySQLComQueryPacketExecutor implements QueryCommandExecutor {
@Getter
private volatile ResponseType responseType;
- private int currentSequenceId;
-
public MySQLComQueryPacketExecutor(final MySQLComQueryPacket packet, final ConnectionSession connectionSession) throws SQLException {
this.connectionSession = connectionSession;
DatabaseType databaseType = DatabaseTypeFactory.getInstance("MySQL");
@@ -101,9 +99,7 @@ public final class MySQLComQueryPacketExecutor implements QueryCommandExecutor {
private Collection<DatabasePacket<?>> processQuery(final QueryResponseHeader queryResponseHeader) {
responseType = ResponseType.QUERY;
- Collection<DatabasePacket<?>> result = ResponsePacketBuilder.buildQueryResponsePackets(queryResponseHeader, characterSet, ServerStatusFlagCalculator.calculateFor(connectionSession));
- currentSequenceId = result.size();
- return result;
+ return ResponsePacketBuilder.buildQueryResponsePackets(queryResponseHeader, characterSet, ServerStatusFlagCalculator.calculateFor(connectionSession));
}
private Collection<DatabasePacket<?>> processUpdate(final UpdateResponseHeader updateResponseHeader) {
@@ -117,7 +113,7 @@ public final class MySQLComQueryPacketExecutor implements QueryCommandExecutor {
@Override
public MySQLPacket getQueryRowPacket() throws SQLException {
- return new MySQLTextResultSetRowPacket(++currentSequenceId, proxyBackendHandler.getRowData().getData());
+ return new MySQLTextResultSetRowPacket(proxyBackendHandler.getRowData().getData());
}
@Override
diff --git a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactory.java b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactory.java
index c70af18c859..338e82cba04 100644
--- a/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactory.java
+++ b/proxy/frontend/mysql/src/main/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactory.java
@@ -40,7 +40,7 @@ public final class MySQLErrPacketFactory {
*/
public static MySQLErrPacket newInstance(final Exception cause) {
SQLException sqlException = SQLExceptionTransformEngine.toSQLException(cause, "MySQL");
- return null == sqlException.getSQLState() ? new MySQLErrPacket(1, MySQLVendorError.ER_INTERNAL_ERROR, getErrorMessage(sqlException)) : createErrPacket(sqlException);
+ return null == sqlException.getSQLState() ? new MySQLErrPacket(MySQLVendorError.ER_INTERNAL_ERROR, getErrorMessage(sqlException)) : createErrPacket(sqlException);
}
private static String getErrorMessage(final SQLException cause) {
@@ -48,6 +48,6 @@ public final class MySQLErrPacketFactory {
}
private static MySQLErrPacket createErrPacket(final SQLException cause) {
- return new MySQLErrPacket(1, cause.getErrorCode(), cause.getSQLState(), cause.getMessage());
+ return new MySQLErrPacket(cause.getErrorCode(), cause.getSQLState(), cause.getMessage());
}
}
diff --git a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/MySQLFrontendEngineTest.java b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/MySQLFrontendEngineTest.java
index a1ee717712a..a6ba5353788 100644
--- a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/MySQLFrontendEngineTest.java
+++ b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/MySQLFrontendEngineTest.java
@@ -98,6 +98,7 @@ public final class MySQLFrontendEngineTest extends ProxyContextRestorer {
when(context.channel()).thenReturn(channel);
when(channel.attr(CommonConstants.CHARSET_ATTRIBUTE_KEY)).thenReturn(mock(Attribute.class));
when(channel.attr(MySQLConstants.MYSQL_CHARACTER_SET_ATTRIBUTE_KEY)).thenReturn(mock(Attribute.class));
+ when(channel.attr(MySQLConstants.MYSQL_SEQUENCE_ID)).thenReturn(mock(Attribute.class));
}
@SneakyThrows(ReflectiveOperationException.class)
diff --git a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngineTest.java b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngineTest.java
index 94c7736b826..b851f5bb038 100644
--- a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngineTest.java
+++ b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/authentication/MySQLAuthenticationEngineTest.java
@@ -200,6 +200,7 @@ public final class MySQLAuthenticationEngineTest extends ProxyContextRestorer {
doReturn(getRemoteAddress()).when(result).remoteAddress();
when(result.attr(CommonConstants.CHARSET_ATTRIBUTE_KEY)).thenReturn(mock(Attribute.class));
when(result.attr(MySQLConstants.MYSQL_CHARACTER_SET_ATTRIBUTE_KEY)).thenReturn(mock(Attribute.class));
+ when(result.attr(MySQLConstants.MYSQL_SEQUENCE_ID)).thenReturn(mock(Attribute.class));
return result;
}
diff --git a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutorTest.java b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutorTest.java
index f1e8f88aaf3..c4a56c274ca 100644
--- a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutorTest.java
+++ b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/command/query/binary/execute/MySQLComStmtExecuteExecutorTest.java
@@ -182,7 +182,6 @@ public final class MySQLComStmtExecuteExecutorTest extends ProxyContextRestorer
assertTrue(mysqlComStmtExecuteExecutor.next());
MySQLPacket actualQueryRowPacket = mysqlComStmtExecuteExecutor.getQueryRowPacket();
assertThat(actualQueryRowPacket, instanceOf(MySQLBinaryResultSetRowPacket.class));
- assertThat(actualQueryRowPacket.getSequenceId(), is(4));
mysqlComStmtExecuteExecutor.close();
verify(proxyBackendHandler).close();
}
diff --git a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactoryTest.java b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactoryTest.java
index eed34f1ab44..0da1d988bc6 100644
--- a/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactoryTest.java
+++ b/proxy/frontend/mysql/src/test/java/org/apache/shardingsphere/proxy/frontend/mysql/err/MySQLErrPacketFactoryTest.java
@@ -34,7 +34,6 @@ public final class MySQLErrPacketFactoryTest {
@Test
public void assertNewInstanceWithSQLExceptionForNullSQLState() {
MySQLErrPacket actual = MySQLErrPacketFactory.newInstance(new SQLException(""));
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(1815));
assertThat(actual.getSqlState(), is(XOpenSQLState.GENERAL_ERROR.getValue()));
assertThat(actual.getErrorMessage(), startsWith("Internal error"));
@@ -43,7 +42,6 @@ public final class MySQLErrPacketFactoryTest {
@Test
public void assertNewInstanceWithSQLException() {
MySQLErrPacket actual = MySQLErrPacketFactory.newInstance(new SQLException("No reason", "XXX", 30000, new RuntimeException("")));
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(30000));
assertThat(actual.getSqlState(), is("XXX"));
assertThat(actual.getErrorMessage(), is("No reason"));
@@ -52,7 +50,6 @@ public final class MySQLErrPacketFactoryTest {
@Test
public void assertNewInstanceWithShardingSphereSQLException() {
MySQLErrPacket actual = MySQLErrPacketFactory.newInstance(new CircuitBreakException());
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(13010));
assertThat(actual.getSqlState(), is(XOpenSQLState.GENERAL_WARNING.getValue()));
assertThat(actual.getErrorMessage(), is("Circuit break open, the request has been ignored."));
@@ -61,7 +58,6 @@ public final class MySQLErrPacketFactoryTest {
@Test
public void assertNewInstanceWithSQLDialectException() {
MySQLErrPacket actual = MySQLErrPacketFactory.newInstance(new UnknownDatabaseException("foo_db"));
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(1049));
assertThat(actual.getSqlState(), is(XOpenSQLState.SYNTAX_ERROR.getValue()));
assertThat(actual.getErrorMessage(), is("Unknown database 'foo_db'"));
@@ -70,7 +66,6 @@ public final class MySQLErrPacketFactoryTest {
@Test
public void assertNewInstanceWithUnknownException() {
MySQLErrPacket actual = MySQLErrPacketFactory.newInstance(new RuntimeException("No reason"));
- assertThat(actual.getSequenceId(), is(1));
assertThat(actual.getErrorCode(), is(30000));
assertThat(actual.getSqlState(), is(XOpenSQLState.GENERAL_ERROR.getValue()));
assertThat(actual.getErrorMessage(), is("Unknown exception: No reason"));