Rename MySQLBaseBinlogEvent (#32563)

* Rename MySQLBaseBinlogEvent * Refactor MySQLWriteRowsBinlogEvent * Refactor MySQLUpdateRowsBinlogEvent * Refactor MySQLDeleteRowsBinlogEvent
apache · Aug 16, 2024 · f0f1c2a · f0f1c2a
1 parent 59455de
commit f0f1c2a
Show file tree

Hide file tree

Showing 13 changed files with 136 additions and 142 deletions.
diff --git a/...tal/binlog/event/AbstractBinlogEvent.java → ...al/binlog/event/MySQLBaseBinlogEvent.java b/...tal/binlog/event/AbstractBinlogEvent.java → ...al/binlog/event/MySQLBaseBinlogEvent.java
@@ -21,11 +21,11 @@
 import lombok.Setter;
 
 /**
- * Abstract binlog event.
+ * MySQL base binlog event.
  */
 @Getter
 @Setter
-public abstract class AbstractBinlogEvent {
+public abstract class MySQLBaseBinlogEvent {
 
     private String fileName;
 

diff --git a/...mental/binlog/event/PlaceholderEvent.java → .../binlog/event/PlaceholderBinlogEvent.java b/...mental/binlog/event/PlaceholderEvent.java → .../binlog/event/PlaceholderBinlogEvent.java
@@ -20,5 +20,5 @@
 /**
  * Placeholder binlog event, unsupported binlog event will replace it into this class.
  */
-public final class PlaceholderEvent extends AbstractBinlogEvent {
+public final class PlaceholderBinlogEvent extends MySQLBaseBinlogEvent {
 }
diff --git a/.../incremental/binlog/event/QueryEvent.java → ...og/event/query/MySQLQueryBinlogEvent.java b/.../incremental/binlog/event/QueryEvent.java → ...og/event/query/MySQLQueryBinlogEvent.java
@@ -15,10 +15,11 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.query;
 
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
 
 /**
  * Query event.This event is written into the binary log file for:
@@ -30,7 +31,7 @@
  */
 @RequiredArgsConstructor
 @Getter
-public final class QueryEvent extends AbstractBinlogEvent {
+public final class MySQLQueryBinlogEvent extends MySQLBaseBinlogEvent {
 
     private final long threadId;
 

diff --git a/...ental/binlog/event/AbstractRowsEvent.java → .../event/rows/MySQLBaseRowsBinlogEvent.java b/...ental/binlog/event/AbstractRowsEvent.java → .../event/rows/MySQLBaseRowsBinlogEvent.java
@@ -15,17 +15,18 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.rows;
 
 import lombok.Getter;
 import lombok.Setter;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
 
 /**
- * Abstract rows event.
+ * MySQL rows base event.
  */
 @Getter
 @Setter
-public abstract class AbstractRowsEvent extends AbstractBinlogEvent {
+public abstract class MySQLBaseRowsBinlogEvent extends MySQLBaseBinlogEvent {
 
     private String databaseName;
 

diff --git a/...emental/binlog/event/DeleteRowsEvent.java → ...vent/rows/MySQLDeleteRowsBinlogEvent.java b/...emental/binlog/event/DeleteRowsEvent.java → ...vent/rows/MySQLDeleteRowsBinlogEvent.java
@@ -15,20 +15,20 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.rows;
 
 import lombok.Getter;
-import lombok.Setter;
+import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
 import java.util.List;
 
 /**
- * Delete rows event.
+ * MySQL delete rows binlog event.
  */
+@RequiredArgsConstructor
 @Getter
-@Setter
-public final class DeleteRowsEvent extends AbstractRowsEvent {
+public final class MySQLDeleteRowsBinlogEvent extends MySQLBaseRowsBinlogEvent {
 
-    private List<Serializable[]> beforeRows;
+    private final List<Serializable[]> beforeRows;
 }
diff --git a/...emental/binlog/event/UpdateRowsEvent.java → ...vent/rows/MySQLUpdateRowsBinlogEvent.java b/...emental/binlog/event/UpdateRowsEvent.java → ...vent/rows/MySQLUpdateRowsBinlogEvent.java
@@ -15,22 +15,22 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.rows;
 
 import lombok.Getter;
-import lombok.Setter;
+import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
 import java.util.List;
 
 /**
- * Update rows event.
+ * MySQL update rows binlog event.
  */
+@RequiredArgsConstructor
 @Getter
-@Setter
-public final class UpdateRowsEvent extends AbstractRowsEvent {
+public final class MySQLUpdateRowsBinlogEvent extends MySQLBaseRowsBinlogEvent {
 
-    private List<Serializable[]> beforeRows;
+    private final List<Serializable[]> beforeRows;
 
-    private List<Serializable[]> afterRows;
+    private final List<Serializable[]> afterRows;
 }
diff --git a/...remental/binlog/event/WriteRowsEvent.java → ...event/rows/MySQLWriteRowsBinlogEvent.java b/...remental/binlog/event/WriteRowsEvent.java → ...event/rows/MySQLWriteRowsBinlogEvent.java
@@ -15,20 +15,20 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.rows;
 
 import lombok.Getter;
-import lombok.Setter;
+import lombok.RequiredArgsConstructor;
 
 import java.io.Serializable;
 import java.util.List;
 
 /**
- * Write rows event.
+ * MySQL write rows binlog event.
  */
+@RequiredArgsConstructor
 @Getter
-@Setter
-public final class WriteRowsEvent extends AbstractRowsEvent {
+public final class MySQLWriteRowsBinlogEvent extends MySQLBaseRowsBinlogEvent {
 
-    private List<Serializable[]> afterRows;
+    private final List<Serializable[]> afterRows;
 }
diff --git a/...st/incremental/binlog/event/XidEvent.java → ...vent/transaction/MySQLXidBinlogEvent.java b/...st/incremental/binlog/event/XidEvent.java → ...vent/transaction/MySQLXidBinlogEvent.java
@@ -15,10 +15,11 @@
  * limitations under the License.
  */
 
-package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event;
+package org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.transaction;
 
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
 
 /**
  * XID event is generated for a COMMIT of a transaction that modifies one or more tables of an XA-capable storage engine.
@@ -27,7 +28,7 @@
  */
 @RequiredArgsConstructor
 @Getter
-public final class XidEvent extends AbstractBinlogEvent {
+public final class MySQLXidBinlogEvent extends MySQLBaseBinlogEvent {
 
     private final long xid;
 }
diff --git a/...pache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java b/...pache/shardingsphere/data/pipeline/mysql/ingest/incremental/client/MySQLBinlogClient.java
@@ -34,8 +34,8 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.shardingsphere.data.pipeline.core.exception.PipelineInternalException;
-import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.AbstractBinlogEvent;
-import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.PlaceholderEvent;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.MySQLBaseBinlogEvent;
+import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.binlog.event.PlaceholderBinlogEvent;
 import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.client.netty.MySQLBinlogEventPacketDecoder;
 import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.client.netty.MySQLCommandPacketDecoder;
 import org.apache.shardingsphere.data.pipeline.mysql.ingest.incremental.client.netty.MySQLNegotiateHandler;
@@ -77,7 +77,7 @@ public final class MySQLBinlogClient {
 
     private final boolean decodeWithTX;
 
-    private final ArrayBlockingQueue<List<AbstractBinlogEvent>> blockingEventQueue = new ArrayBlockingQueue<>(2500);
+    private final ArrayBlockingQueue<List<MySQLBaseBinlogEvent>> blockingEventQueue = new ArrayBlockingQueue<>(2500);
 
     private EventLoopGroup eventLoopGroup;
 
@@ -226,8 +226,8 @@ private void dumpBinlog(final String binlogFileName, final long binlogPosition,
         channel.writeAndFlush(new MySQLComBinlogDumpCommandPacket((int) binlogPosition, connectInfo.getServerId(), binlogFileName));
     }
 
-    private AbstractBinlogEvent getLastBinlogEvent(final String binlogFileName, final long binlogPosition) {
-        PlaceholderEvent result = new PlaceholderEvent();
+    private MySQLBaseBinlogEvent getLastBinlogEvent(final String binlogFileName, final long binlogPosition) {
+        PlaceholderBinlogEvent result = new PlaceholderBinlogEvent();
         result.setFileName(binlogFileName);
         result.setPosition(binlogPosition);
         return result;
@@ -242,12 +242,12 @@ private void resetSequenceID() {
      *
      * @return binlog event
      */
-    public synchronized List<AbstractBinlogEvent> poll() {
+    public synchronized List<MySQLBaseBinlogEvent> poll() {
         if (!running) {
             return Collections.emptyList();
         }
         try {
-            List<AbstractBinlogEvent> result = blockingEventQueue.poll(100L, TimeUnit.MILLISECONDS);
+            List<MySQLBaseBinlogEvent> result = blockingEventQueue.poll(100L, TimeUnit.MILLISECONDS);
             return null == result ? Collections.emptyList() : result;
         } catch (final InterruptedException ignored) {
             Thread.currentThread().interrupt();
@@ -314,11 +314,11 @@ public void exceptionCaught(final ChannelHandlerContext ctx, final Throwable cau
 
     private final class MySQLBinlogEventHandler extends ChannelInboundHandlerAdapter {
 
-        private final AtomicReference<AbstractBinlogEvent> lastBinlogEvent;
+        private final AtomicReference<MySQLBaseBinlogEvent> lastBinlogEvent;
 
         private final AtomicBoolean reconnectRequested = new AtomicBoolean(false);
 
-        MySQLBinlogEventHandler(final AbstractBinlogEvent lastBinlogEvent) {
+        MySQLBinlogEventHandler(final MySQLBaseBinlogEvent lastBinlogEvent) {
             this.lastBinlogEvent = new AtomicReference<>(lastBinlogEvent);
         }
 
@@ -329,7 +329,7 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) throw
                 return;
             }
             if (msg instanceof List) {
-                List<AbstractBinlogEvent> records = (List<AbstractBinlogEvent>) msg;
+                List<MySQLBaseBinlogEvent> records = (List<MySQLBaseBinlogEvent>) msg;
                 if (records.isEmpty()) {
                     log.warn("The records is empty");
                     return;
@@ -338,8 +338,8 @@ public void channelRead(final ChannelHandlerContext ctx, final Object msg) throw
                 blockingEventQueue.put(records);
                 return;
             }
-            if (msg instanceof AbstractBinlogEvent) {
-                lastBinlogEvent.set((AbstractBinlogEvent) msg);
+            if (msg instanceof MySQLBaseBinlogEvent) {
+                lastBinlogEvent.set((MySQLBaseBinlogEvent) msg);
                 blockingEventQueue.put(Collections.singletonList(lastBinlogEvent.get()));
             }
         }