diff --git a/.github/workflows/pytest.yml b/.github/workflows/pytest.yml
index 265c4d57..47ea5f40 100644
--- a/.github/workflows/pytest.yml
+++ b/.github/workflows/pytest.yml
@@ -28,9 +28,17 @@ jobs:
         run: |
           docker compose create
           docker compose start
-          # Wait for the services to accept connections,
-          # TODO: do that smarter, poll connection attempt until it succeeds
-          sleep 30
+          echo "wait mysql server"
+        
+          while :
+          do
+            if mysql -h 127.0.0.1 --user=root --execute "SELECT version();" 2>&1 >/dev/null && mysql -h 127.0.0.1 --port=3307  --user=root --execute "SELECT version();" 2>&1 >/dev/null; then
+              break
+            fi
+            sleep 1
+          done
+          
+          echo "run pytest"
 
       - name: Install dependencies
         run: |
diff --git a/.mariadb/my.cnf b/.mariadb/my.cnf
new file mode 100644
index 00000000..c530c80c
--- /dev/null
+++ b/.mariadb/my.cnf
@@ -0,0 +1,23 @@
+[client-server]
+# Port or socket location where to connect
+# port = 3306
+socket = /run/mysqld/mysqld.sock
+
+# Import all .cnf files from configuration directory
+
+!includedir /etc/mysql/mariadb.conf.d/
+!includedir /etc/mysql/conf.d/
+
+
+[mariadb]
+plugin_load_add = file_key_management
+# Key files that are not encrypted
+loose_file_key_management_filename = /opt/key_file/no_encryption_key.key
+
+# Encrypted key file
+# loose_file_key_management_filename=/opt/key_file/keyfile.enc
+# loose_file_key_management_filekey=FILE:/opt/key_file/no_encryption_key.key
+# file_key_management_encryption_algorithm=aes_ctr
+
+# Set encrypt_binlog
+encrypt_binlog=ON
\ No newline at end of file
diff --git a/.mariadb/no_encryption_key.key b/.mariadb/no_encryption_key.key
new file mode 100755
index 00000000..476ede79
--- /dev/null
+++ b/.mariadb/no_encryption_key.key
@@ -0,0 +1 @@
+1;dda0ccb18a28b0b4c2448b5f0217a134
\ No newline at end of file
diff --git a/README.md b/README.md
index 78cf85a1..2bc28db5 100644
--- a/README.md
+++ b/README.md
@@ -4,7 +4,7 @@ python-mysql-replication
 <a href="https://travis-ci.org/noplay/python-mysql-replication"><img src="https://travis-ci.org/noplay/python-mysql-replication.svg?branch=master"></a>&nbsp;
 <a href="https://pypi.python.org/pypi/mysql-replication"><img src="http://img.shields.io/pypi/dm/mysql-replication.svg"></a>
 
-Pure Python Implementation of MySQL replication protocol build on top of PyMYSQL. This allow you to receive event like insert, update, delete with their datas and raw SQL queries.
+Pure Python Implementation of MySQL replication protocol build on top of PyMYSQL. This allows you to receive event like insert, update, delete with their datas and raw SQL queries.
 
 Use cases
 ===========
@@ -56,6 +56,11 @@ Limitations
 
 https://python-mysql-replication.readthedocs.org/en/latest/limitations.html
 
+Featured Books
+=============
+
+[Data Pipelines Pocket Reference](https://www.oreilly.com/library/view/data-pipelines-pocket/9781492087823/) (by James Densmore, O'Reilly): Introduced and exemplified in Chapter 4: Data Ingestion: Extracting Data.
+
 Projects using this library
 ===========================
 
diff --git a/docker-compose-test.yml b/docker-compose-test.yml
new file mode 100644
index 00000000..38d95827
--- /dev/null
+++ b/docker-compose-test.yml
@@ -0,0 +1,59 @@
+version: '3.2'
+services:
+  percona-5.7:
+    platform: linux/amd64
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+      MYSQL_DATABASE: pymysqlreplication_test
+    ports:
+      - 3306:3306
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates
+    restart: always
+    networks:
+      - default
+
+  percona-5.7-ctl:
+    image: percona:5.7
+    environment:
+      MYSQL_ALLOW_EMPTY_PASSWORD: true
+      MYSQL_DATABASE: pymysqlreplication_test
+    ports:
+      - 3307:3307
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
+
+  pymysqlreplication:
+    build:
+      context: .
+      dockerfile: test.Dockerfile
+      args:
+        BASE_IMAGE: python:3.11-alpine
+        MYSQL_5_7: percona-5.7
+        MYSQL_5_7_CTL: percona-5.7-ctl
+
+    command:
+      - /bin/sh
+      - -ce
+      - |
+        echo "wait mysql server"
+        
+        while :
+        do
+          if mysql -h percona-5.7 --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null && mysql -h percona-5.7-ctl --port=3307  --user=root --execute "USE pymysqlreplication_test;" 2>&1 >/dev/null; then
+            break
+          fi
+          sleep 1
+        done
+        
+        echo "run pytest"
+        pytest -k "not test_no_trailing_rotate_event and not test_end_log_pos"
+
+    working_dir: /pymysqlreplication
+    networks:
+      - default
+    depends_on:
+      - percona-5.7
+      - percona-5.7-ctl
+
+networks:
+  default: {}
diff --git a/docker-compose.yml b/docker-compose.yml
index 5032ae6e..3de0a9cf 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -16,10 +16,30 @@ services:
       - 3307:3307
     command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3307
 
+  mariadb-10.6:
+    image: mariadb:10.6
+    environment:
+      MARIADB_ALLOW_EMPTY_ROOT_PASSWORD: 1
+    ports:
+      - "3308:3306"
+    command: |
+      --server-id=1
+      --default-authentication-plugin=mysql_native_password
+      --log-bin=master-bin
+      --binlog-format=row
+      --log-slave-updates=on
+    volumes:
+      - type: bind
+        source: ./.mariadb
+        target: /opt/key_file
+      - type: bind
+        source: ./.mariadb/my.cnf
+        target: /etc/mysql/my.cnf  
+
   percona-8.0:
     image: percona:8.0
     environment:
       MYSQL_ALLOW_EMPTY_PASSWORD: true
     ports:
-      - 3308:3308
-    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3308
+      - 3309:3309
+    command: mysqld --log-bin=mysql-bin.log --server-id 1 --binlog-format=row --gtid_mode=on --enforce-gtid-consistency=on --log_slave_updates -P 3309
diff --git a/docs/developement.rst b/docs/developement.rst
index 30e257e0..68eecc21 100644
--- a/docs/developement.rst
+++ b/docs/developement.rst
@@ -23,7 +23,7 @@ When it's possible we have an unit test.
 *pymysqlreplication/tests/* contains the test suite. The test suite
 use the standard *unittest* Python module.
 
-**Be carefull** tests will reset the binary log of your MySQL server.
+**Be careful** tests will reset the binary log of your MySQL server.
 
 Make sure you have the following configuration set in your mysql config file (usually my.cnf on development env):
 
diff --git a/examples/mariadb_gtid/read_event.py b/examples/mariadb_gtid/read_event.py
index cc88a97f..49598c3f 100644
--- a/examples/mariadb_gtid/read_event.py
+++ b/examples/mariadb_gtid/read_event.py
@@ -1,7 +1,7 @@
 import pymysql
 
 from pymysqlreplication import BinLogStreamReader, gtid
-from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent
+from pymysqlreplication.event import GtidEvent, RotateEvent, MariadbGtidEvent, QueryEvent,MariadbAnnotateRowsEvent
 from pymysqlreplication.row_event import WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent
 
 MARIADB_SETTINGS = {
@@ -65,10 +65,12 @@ def query_server_id(self):
             RotateEvent,
             WriteRowsEvent,
             UpdateRowsEvent,
-            DeleteRowsEvent
+            DeleteRowsEvent,
+            MariadbAnnotateRowsEvent
         ],
         auto_position=gtid,
-        is_mariadb=True
+        is_mariadb=True,
+        annotate_rows_event=True
     )
 
     print('Starting reading events from GTID ', gtid)
diff --git a/pymysqlreplication/binlogstream.py b/pymysqlreplication/binlogstream.py
index fa65aa22..9c324a06 100644
--- a/pymysqlreplication/binlogstream.py
+++ b/pymysqlreplication/binlogstream.py
@@ -14,7 +14,8 @@
     QueryEvent, RotateEvent, FormatDescriptionEvent,
     XidEvent, GtidEvent, StopEvent, XAPrepareEvent,
     BeginLoadQueryEvent, ExecuteLoadQueryEvent,
-    HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent)
+    HeartbeatLogEvent, NotImplementedEvent, MariadbGtidEvent,
+    MariadbAnnotateRowsEvent, RandEvent, MariadbStartEncryptionEvent)
 from .exceptions import BinLogNotEnabled
 from .row_event import (
     UpdateRowsEvent, WriteRowsEvent, DeleteRowsEvent, TableMapEvent)
@@ -141,6 +142,7 @@ def __init__(self, connection_settings, server_id,
                  fail_on_table_metadata_unavailable=False,
                  slave_heartbeat=None,
                  is_mariadb=False,
+                 annotate_rows_event=False,
                  ignore_decode_errors=False):
         """
         Attributes:
@@ -166,7 +168,8 @@ def __init__(self, connection_settings, server_id,
             skip_to_timestamp: Ignore all events until reaching specified
                                timestamp.
             report_slave: Report slave in SHOW SLAVE HOSTS.
-            slave_uuid: Report slave_uuid in SHOW SLAVE HOSTS.
+            slave_uuid: Report slave_uuid or replica_uuid in SHOW SLAVE HOSTS(MySQL 8.0.21-) or
+                        SHOW REPLICAS(MySQL 8.0.22+) depends on your MySQL version.
             fail_on_table_metadata_unavailable: Should raise exception if we
                                                 can't get table information on
                                                 row_events
@@ -178,6 +181,8 @@ def __init__(self, connection_settings, server_id,
                              for semantics
             is_mariadb: Flag to indicate it's a MariaDB server, used with auto_position
                     to point to Mariadb specific GTID.
+            annotate_rows_event: Parameter value to enable annotate rows event in mariadb,
+                    used with 'is_mariadb'
             ignore_decode_errors: If true, any decode errors encountered 
                                   when reading column data will be ignored.
         """
@@ -219,6 +224,7 @@ def __init__(self, connection_settings, server_id,
         self.auto_position = auto_position
         self.skip_to_timestamp = skip_to_timestamp
         self.is_mariadb = is_mariadb
+        self.__annotate_rows_event = annotate_rows_event
 
         if end_log_pos:
             self.is_past_end_log_pos = False
@@ -301,7 +307,7 @@ def __connect_to_stream(self):
 
         if self.slave_uuid:
             cur = self._stream_connection.cursor()
-            cur.execute("set @slave_uuid= '%s'" % self.slave_uuid)
+            cur.execute("SET @slave_uuid = %s, @replica_uuid = %s", (self.slave_uuid, self.slave_uuid))
             cur.close()
 
         if self.slave_heartbeat:
@@ -331,67 +337,39 @@ def __connect_to_stream(self):
         self._register_slave()
 
         if not self.auto_position:
-            # only when log_file and log_pos both provided, the position info is
-            # valid, if not, get the current position from master
-            if self.log_file is None or self.log_pos is None:
-                cur = self._stream_connection.cursor()
-                cur.execute("SHOW MASTER STATUS")
-                master_status = cur.fetchone()
-                if master_status is None:
-                    raise BinLogNotEnabled()
-                self.log_file, self.log_pos = master_status[:2]
-                cur.close()
-
-            prelude = struct.pack('<i', len(self.log_file) + 11) \
-                + bytes(bytearray([COM_BINLOG_DUMP]))
-
-            if self.__resume_stream:
-                prelude += struct.pack('<I', self.log_pos)
-            else:
-                prelude += struct.pack('<I', 4)
-
-            flags = 0
-            if not self.__blocking:
-                flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
-            prelude += struct.pack('<H', flags)
-
-            prelude += struct.pack('<I', self.__server_id)
-            prelude += self.log_file.encode()
-        else:
             if self.is_mariadb:
-                # https://mariadb.com/kb/en/5-slave-registration/
-                cur = self._stream_connection.cursor()
-                cur.execute("SET @slave_connect_state='%s'" % self.auto_position)
-                cur.execute("SET @slave_gtid_strict_mode=1")
-                cur.execute("SET @slave_gtid_ignore_duplicates=0")
-                cur.close()
-
-                # https://mariadb.com/kb/en/com_binlog_dump/
-                header_size = (
-                        4 +  # binlog pos
-                        2 +  # binlog flags
-                        4 +  # slave server_id,
-                        4    # requested binlog file name , set it to empty
-                )
-
-                prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
-
-                # binlog pos
-                prelude += struct.pack('<i', 4)
+                prelude = self.__set_mariadb_settings()
+            else:
+                # only when log_file and log_pos both provided, the position info is
+                # valid, if not, get the current position from master
+                if self.log_file is None or self.log_pos is None:
+                    cur = self._stream_connection.cursor()
+                    cur.execute("SHOW MASTER STATUS")
+                    master_status = cur.fetchone()
+                    if master_status is None:
+                        raise BinLogNotEnabled()
+                    self.log_file, self.log_pos = master_status[:2]
+                    cur.close()
+
+                prelude = struct.pack('<i', len(self.log_file) + 11) \
+                    + bytes(bytearray([COM_BINLOG_DUMP]))
+
+                if self.__resume_stream:
+                    prelude += struct.pack('<I', self.log_pos)
+                else:
+                    prelude += struct.pack('<I', 4)
 
                 flags = 0
+
                 if not self.__blocking:
                     flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
-                
-                # binlog flags
                 prelude += struct.pack('<H', flags)
 
-                # server id (4 bytes)
                 prelude += struct.pack('<I', self.__server_id)
-
-                # empty_binlog_name (4 bytes)
-                prelude += b'\0\0\0\0'
-                
+                prelude += self.log_file.encode()
+        else:
+            if self.is_mariadb:
+                prelude = self.__set_mariadb_settings()        
             else:
                 # Format for mysql packet master_auto_position
                 #
@@ -473,6 +451,48 @@ def __connect_to_stream(self):
             self._stream_connection._next_seq_id = 1
         self.__connected_stream = True
 
+    def __set_mariadb_settings(self):
+        # https://mariadb.com/kb/en/5-slave-registration/
+        cur = self._stream_connection.cursor()
+        if self.auto_position != None : 
+            cur.execute("SET @slave_connect_state='%s'" % self.auto_position)
+        cur.execute("SET @slave_gtid_strict_mode=1")
+        cur.execute("SET @slave_gtid_ignore_duplicates=0")
+        cur.close()
+
+        # https://mariadb.com/kb/en/com_binlog_dump/
+        header_size = (
+                4 +  # binlog pos
+                2 +  # binlog flags
+                4 +  # slave server_id,
+                4    # requested binlog file name , set it to empty
+        )
+
+        prelude = struct.pack('<i', header_size) + bytes(bytearray([COM_BINLOG_DUMP]))
+
+        # binlog pos
+        prelude += struct.pack('<i', 4)
+
+        flags = 0
+
+        # Enable annotate rows event 
+        if self.__annotate_rows_event:
+            flags |= 0x02 # BINLOG_SEND_ANNOTATE_ROWS_EVENT
+
+        if not self.__blocking:
+            flags |= 0x01  # BINLOG_DUMP_NON_BLOCK
+        
+        # binlog flags
+        prelude += struct.pack('<H', flags)
+
+        # server id (4 bytes)
+        prelude += struct.pack('<I', self.__server_id)
+
+        # empty_binlog_name (4 bytes)
+        prelude += b'\0\0\0\0'
+
+        return prelude
+
     def fetchone(self):
         while True:
             if self.end_log_pos and self.is_past_end_log_pos:
@@ -600,7 +620,10 @@ def _allowed_event_list(self, only_events, ignored_events,
                 TableMapEvent,
                 HeartbeatLogEvent,
                 NotImplementedEvent,
-                MariadbGtidEvent
+                MariadbGtidEvent,
+                MariadbAnnotateRowsEvent,
+                RandEvent,
+                MariadbStartEncryptionEvent
                 ))
         if ignored_events is not None:
             for e in ignored_events:
diff --git a/pymysqlreplication/bitmap.py b/pymysqlreplication/bitmap.py
index 740894ac..fd2096b6 100644
--- a/pymysqlreplication/bitmap.py
+++ b/pymysqlreplication/bitmap.py
@@ -19,7 +19,7 @@
     4, 5, 5, 6, 5, 6, 6, 7, 5, 6, 6, 7, 6, 7, 7, 8,
 ]
 
-# Calculate totol bit counts in a bitmap
+# Calculate total bit counts in a bitmap
 def BitCount(bitmap):
     n = 0
     for i in range(0, len(bitmap)):
diff --git a/pymysqlreplication/event.py b/pymysqlreplication/event.py
index d75c9db8..db8c6543 100644
--- a/pymysqlreplication/event.py
+++ b/pymysqlreplication/event.py
@@ -111,6 +111,24 @@ def _dump(self):
         print('GTID:', self.gtid)
 
 
+class MariadbAnnotateRowsEvent(BinLogEvent):
+    """
+    Annotate rows event 
+    If you want to check this binlog, change the value of the flag(line 382 of the 'binlogstream.py') option to 2 
+    https://mariadb.com/kb/en/annotate_rows_event/
+
+    Attributes:
+        sql_statement: The SQL statement
+    """
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(MariadbAnnotateRowsEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+        self.sql_statement = self.packet.read(event_size)
+
+    def _dump(self):
+        super(MariadbAnnotateRowsEvent, self)._dump()
+        print("SQL statement :", self.sql_statement)   
+
+
 class RotateEvent(BinLogEvent):
     """Change MySQL bin log file
 
@@ -133,7 +151,7 @@ def dump(self):
 
 class XAPrepareEvent(BinLogEvent):
     """An XA prepare event is generated for a XA prepared transaction.
-    Like Xid_event it contans XID of the *prepared* transaction
+    Like Xid_event it contains XID of the *prepared* transaction
 
     Attributes:
         one_phase: current XA transaction commit method
@@ -244,7 +262,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
         # Payload
         status_vars_end_pos = self.packet.read_bytes + self.status_vars_length
-        while self.packet.read_bytes < status_vars_end_pos: # while 남은 data length가 얼마만큼? OR read_bytes
+        while self.packet.read_bytes < status_vars_end_pos:
             # read KEY for status variable
             status_vars_key = self.packet.read_uint8()
             # read VALUE for status variable
@@ -314,7 +332,7 @@ def _read_status_vars_value_for_key(self, key):
             """
             mts_accessed_dbs < 254:
                 `mts_accessed_dbs` is equal to the number of dbs
-                acessed by the query event.
+                accessed by the query event.
             mts_accessed_dbs == 254:
                 This is the case where the number of dbs accessed
                 is 1 and the name of the only db is ""
@@ -436,6 +454,68 @@ def _dump(self):
         print("type: %d" % (self.type))
         print("Value: %d" % (self.value))
 
+class RandEvent(BinLogEvent):
+    """
+    RandEvent is generated every time a statement uses the RAND() function.
+    Indicates the seed values to use for generating a random number with RAND() in the next statement.
+
+    RandEvent only works in statement-based logging (need to set binlog_format as 'STATEMENT')
+    and only works when the seed number is not specified.
+
+    :ivar seed1: int - value for the first seed
+    :ivar seed2: int - value for the second seed
+    """
+    
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(RandEvent, self).__init__(from_packet, event_size, table_map,
+                                        ctl_connection, **kwargs)
+        # Payload
+        self._seed1 = self.packet.read_uint64()
+        self._seed2 = self.packet.read_uint64()
+
+    @property
+    def seed1(self):
+        """Get the first seed value"""
+        return self._seed1
+
+    @property
+    def seed2(self):
+        """Get the second seed value"""
+        return self._seed2
+
+    def _dump(self):
+        super(RandEvent, self)._dump()
+        print("seed1: %d" % (self.seed1))
+        print("seed2: %d" % (self.seed2))
+
+class MariadbStartEncryptionEvent(BinLogEvent):
+    """
+    Since MariaDB 10.1.7, 
+    the START_ENCRYPTION event is written to every binary log file 
+    if encrypt_binlog is set to ON. Prior to enabling this setting, 
+    additional configuration steps are required in MariaDB. 
+    (Link: https://mariadb.com/kb/en/encrypting-binary-logs/)
+
+    This event is written just once, after the Format Description event
+
+    Attributes:
+        schema: The Encryption scheme, always set to 1 for system files.
+        key_version: The Encryption key version.
+        nonce: Nonce (12 random bytes) of current binlog file.
+    """
+
+    def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
+        super(MariadbStartEncryptionEvent, self).__init__(from_packet, event_size, table_map, ctl_connection, **kwargs)
+
+        self.schema = self.packet.read_uint8()
+        self.key_version = self.packet.read_uint32()
+        self.nonce = self.packet.read(12)
+
+    def _dump(self):
+        print("Schema: %d" % self.schema)
+        print("Key version: %d" % self.key_version)
+        print(f"Nonce: {self.nonce}")
+
 
 class NotImplementedEvent(BinLogEvent):
     def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs):
diff --git a/pymysqlreplication/packet.py b/pymysqlreplication/packet.py
index 317c9d0d..b48ed721 100644
--- a/pymysqlreplication/packet.py
+++ b/pymysqlreplication/packet.py
@@ -70,6 +70,7 @@ class BinLogPacketWrapper(object):
         constants.EXECUTE_LOAD_QUERY_EVENT: event.ExecuteLoadQueryEvent,
         constants.HEARTBEAT_LOG_EVENT: event.HeartbeatLogEvent,
         constants.XA_PREPARE_EVENT: event.XAPrepareEvent,
+        constants.RAND_EVENT: event.RandEvent,
         # row_event
         constants.UPDATE_ROWS_EVENT_V1: row_event.UpdateRowsEvent,
         constants.WRITE_ROWS_EVENT_V1: row_event.WriteRowsEvent,
@@ -83,11 +84,11 @@ class BinLogPacketWrapper(object):
         constants.ANONYMOUS_GTID_LOG_EVENT: event.NotImplementedEvent,
         constants.PREVIOUS_GTIDS_LOG_EVENT: event.NotImplementedEvent,
         # MariaDB GTID
-        constants.MARIADB_ANNOTATE_ROWS_EVENT: event.NotImplementedEvent,
+        constants.MARIADB_ANNOTATE_ROWS_EVENT: event.MariadbAnnotateRowsEvent,
         constants.MARIADB_BINLOG_CHECKPOINT_EVENT: event.NotImplementedEvent,
         constants.MARIADB_GTID_EVENT: event.MariadbGtidEvent,
         constants.MARIADB_GTID_GTID_LIST_EVENT: event.NotImplementedEvent,
-        constants.MARIADB_START_ENCRYPTION_EVENT: event.NotImplementedEvent
+        constants.MARIADB_START_ENCRYPTION_EVENT: event.MariadbStartEncryptionEvent
     }
 
     def __init__(self, from_packet, table_map,
diff --git a/pymysqlreplication/row_event.py b/pymysqlreplication/row_event.py
index 57219318..b45c2f1c 100644
--- a/pymysqlreplication/row_event.py
+++ b/pymysqlreplication/row_event.py
@@ -676,7 +676,7 @@ def __init__(self, from_packet, event_size, table_map, ctl_connection, **kwargs)
 
                     ordinal_pos_loc += 1
                 except IndexError:
-                    # this a dirty hack to prevent row events containing columns which have been dropped prior
+                    # this is a dirty hack to prevent row events containing columns which have been dropped prior
                     # to pymysqlreplication start, but replayed from binlog from blowing up the service.
                     # TODO: this does not address the issue if the column other than the last one is dropped
                     column_schema = {
diff --git a/pymysqlreplication/tests/base.py b/pymysqlreplication/tests/base.py
index b9bd03cc..5e1c36b4 100644
--- a/pymysqlreplication/tests/base.py
+++ b/pymysqlreplication/tests/base.py
@@ -19,11 +19,9 @@ def ignoredEvents(self):
         return []
 
     def setUp(self):
-
-        db = os.environ.get('DB')
         # default
         self.database = {
-            "host": "localhost",
+            "host": os.environ.get("MYSQL_5_7") or "localhost",
             "user": "root",
             "passwd": "",
             "port": 3306,
@@ -128,3 +126,29 @@ def bin_log_basename(self):
         bin_log_basename = cursor.fetchone()[0]
         bin_log_basename = bin_log_basename.split("/")[-1]
         return bin_log_basename
+
+
+class PyMySQLReplicationMariaDbTestCase(PyMySQLReplicationTestCase):
+    def setUp(self):
+        # default
+        self.database = {
+            "host": "localhost",
+            "user": "root",
+            "passwd": "",
+            "port": 3308,
+            "use_unicode": True,
+            "charset": "utf8",
+            "db": "pymysqlreplication_test"
+        }
+
+        self.conn_control = None
+        db = copy.copy(self.database)
+        db["db"] = None
+        self.connect_conn_control(db)
+        self.execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
+        self.execute("CREATE DATABASE pymysqlreplication_test")
+        db = copy.copy(self.database)
+        self.connect_conn_control(db)
+        self.stream = None
+        self.resetBinLog()
+        
\ No newline at end of file
diff --git a/pymysqlreplication/tests/test_basic.py b/pymysqlreplication/tests/test_basic.py
index 18918586..c40a83ce 100644
--- a/pymysqlreplication/tests/test_basic.py
+++ b/pymysqlreplication/tests/test_basic.py
@@ -1,4 +1,6 @@
 # -*- coding: utf-8 -*-
+import os
+
 import pymysql
 import copy
 import time
@@ -17,19 +19,21 @@
 from pymysqlreplication.exceptions import TableMetadataUnavailableError
 from pymysqlreplication.constants.BINLOG import *
 from pymysqlreplication.row_event import *
+from pathlib import Path
 
-__all__ = ["TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings",
-           "TestGtidBinLogStreamReader", "TestOptionalMetaData"]
-
+__all__ = [
+    "TestBasicBinLogStreamReader", "TestMultipleRowBinLogStreamReader", "TestCTLConnectionSettings",
+    "TestGtidBinLogStreamReader", "TestMariadbBinlogStreamReader", "TestStatementConnectionSetting", "TestOptionalMetaData"
+]
 
 class TestBasicBinLogStreamReader(base.PyMySQLReplicationTestCase):
     def ignoredEvents(self):
         return [GtidEvent]
 
     def test_allowed_event_list(self):
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 16)
-        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 15)
-        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 15)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, False)), 19)
+        self.assertEqual(len(self.stream._allowed_event_list(None, None, True)), 18)
+        self.assertEqual(len(self.stream._allowed_event_list(None, [RotateEvent], False)), 18)
         self.assertEqual(len(self.stream._allowed_event_list([RotateEvent], None, False)), 1)
 
     def test_read_query_event(self):
@@ -770,6 +774,8 @@ def setUp(self):
         ctl_db = copy.copy(self.database)
         ctl_db["db"] = None
         ctl_db["port"] = 3307
+        if os.environ.get("MYSQL_5_7_CTL") is not None:
+            ctl_db["host"] = os.environ.get("MYSQL_5_7_CTL")
         self.ctl_conn_control = pymysql.connect(**ctl_db)
         self.ctl_conn_control.cursor().execute("DROP DATABASE IF EXISTS pymysqlreplication_test")
         self.ctl_conn_control.cursor().execute("CREATE DATABASE pymysqlreplication_test")
@@ -1006,6 +1012,103 @@ def test_parsing(self):
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac:1-:1")
             gtid = Gtid("57b70f4e-20d3-11e5-a393-4a63946f7eac::1")
 
+class TestMariadbBinlogStreamReader(base.PyMySQLReplicationMariaDbTestCase):
+    
+    def test_annotate_rows_event(self):
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        # Insert first event
+        query = "BEGIN;"
+        self.execute(query)
+        insert_query = b"INSERT INTO test (id, data) VALUES(1, 'Hello')"
+        self.execute(insert_query)
+        query = "COMMIT;"
+        self.execute(query)
+
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database, 
+            server_id=1024, 
+            blocking=False,
+            only_events=[MariadbAnnotateRowsEvent],
+            is_mariadb=True,
+            annotate_rows_event=True,
+            )
+        
+        event = self.stream.fetchone()
+        #Check event type 160,MariadbAnnotateRowsEvent
+        self.assertEqual(event.event_type,160)
+        #Check self.sql_statement
+        self.assertEqual(event.sql_statement,insert_query)
+        self.assertIsInstance(event,MariadbAnnotateRowsEvent)
+        
+    def test_start_encryption_event(self):
+        query = "CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data VARCHAR (50) NOT NULL, PRIMARY KEY (id))"
+        self.execute(query)
+        query = "INSERT INTO test (data) VALUES('Hello World')"
+        self.execute(query)
+        self.execute("COMMIT")
+
+        self.assertIsInstance(self.stream.fetchone(), RotateEvent)
+        self.assertIsInstance(self.stream.fetchone(), FormatDescriptionEvent)
+
+        start_encryption_event = self.stream.fetchone()
+        self.assertIsInstance(start_encryption_event, MariadbStartEncryptionEvent)
+
+        schema = start_encryption_event.schema
+        key_version = start_encryption_event.key_version
+        nonce = start_encryption_event.nonce
+
+        from pathlib import Path
+
+        encryption_key_file_path = Path(__file__).parent.parent.parent
+
+        try:
+            with open(f"{encryption_key_file_path}/.mariadb/no_encryption_key.key", "r") as key_file:
+                first_line = key_file.readline()
+                key_version_from_key_file = int(first_line.split(";")[0])
+        except Exception as e:
+            self.fail("raised unexpected exception: {exception}".format(exception=e))
+        finally:
+            self.resetBinLog()
+
+        # schema is always 1
+        self.assertEqual(schema, 1)
+        self.assertEqual(key_version, key_version_from_key_file)
+        self.assertEqual(type(nonce), bytes)
+        self.assertEqual(len(nonce), 12)        
+        
+class TestStatementConnectionSetting(base.PyMySQLReplicationTestCase):
+    def setUp(self):
+        super(TestStatementConnectionSetting, self).setUp()
+        self.stream.close()
+        self.stream = BinLogStreamReader(
+            self.database,
+            server_id=1024,
+            only_events=(RandEvent, QueryEvent),
+            fail_on_table_metadata_unavailable=True
+        )
+        self.execute("SET @@binlog_format='STATEMENT'")
+
+    def test_rand_event(self):
+        self.execute("CREATE TABLE test (id INT NOT NULL AUTO_INCREMENT, data INT NOT NULL, PRIMARY KEY (id))")
+        self.execute("INSERT INTO test (data) VALUES(RAND())")
+        self.execute("COMMIT")
+
+        self.assertEqual(self.bin_log_format(), "STATEMENT")
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+        self.assertIsInstance(self.stream.fetchone(), QueryEvent)
+
+        expect_rand_event = self.stream.fetchone()
+        self.assertIsInstance(expect_rand_event, RandEvent)
+        self.assertEqual(type(expect_rand_event.seed1), int)
+        self.assertEqual(type(expect_rand_event.seed2), int)
+
+    def tearDown(self):
+        self.execute("SET @@binlog_format='ROW'")
+        self.assertEqual(self.bin_log_format(), "ROW")
+        super(TestStatementConnectionSetting, self).tearDown()        
+
 
 class TestOptionalMetaData(base.PyMySQLReplicationTestCase):
     def setUp(self):
diff --git a/test.Dockerfile b/test.Dockerfile
new file mode 100644
index 00000000..6aa2e34a
--- /dev/null
+++ b/test.Dockerfile
@@ -0,0 +1,16 @@
+ARG BASE_IMAGE=${BASE_IMAGE:-python:3.11-alpine}
+FROM ${BASE_IMAGE}
+
+COPY pymysqlreplication pymysqlreplication
+COPY README.md README.md
+COPY setup.py setup.py
+RUN apk add bind-tools
+RUN apk add mysql-client
+RUN pip install .
+RUN pip install pytest
+
+ARG MYSQL_5_7
+ENV MYSQL_5_7 ${MYSQL_5_7}
+
+ARG MYSQL_5_7_CTL
+ENV MYSQL_5_7_CTL ${MYSQL_5_7_CTL}
\ No newline at end of file