<dependency>
    <groupId>org.mariadb</groupId>
    <artifactId>r2dbc-mariadb</artifactId>
    <version>1.4.1</version>
</dependency>

CREATE USER 'connr2dbc_test'@'192.0.2.50'
   IDENTIFIED BY 'db_user_password';

GRANT ALL PRIVILEGES
   ON test.*
   TO 'connr2dbc_test'@'192.0.2.50';

$ export CLASSPATH="/path/to/application:/path/to/r2dbc-mariadb-1.4.1.jar"

$ javac App.java

$ java App

GRANT ALL PRIVILEGES
   ON test.*
   TO 'connr2dbc_test'@'192.0.2.50';

// Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;

// Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   // Main Process
   public static void main(String[] args) {
      //Initialize Connection Factory
      initConnectionFactory();

     //Initialize a Connection
     conn = connFactory.create().block();

      // Use the connection
      //conn.
   }

   public static void initConnectionFactory() {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      }
   }
}

// Imports the @Id annotation type, which demarcates an identifier.
import org.springframework.data.annotation.Id;

// This is an Entity class
// It has the same name as the text.contact table
public class Contact {

      // The class members correspond to columns
      // in the test.contact table
      private int id;
      private String first_name;
      private String last_name;
      private String email;

      // Constructor
      public Contact(int id, String first_name, String last_name, String email) {
         this.id = id;
         this.first_name = first_name;
         this.last_name = last_name;
         this.email = email;
      }

      // The @Id annotation indicates that this field
      // is the primary key column
      @Id
      public int getId() {
         return id;
      }

      public String getFirst_name() {
         return first_name;
      }

      public String getLast_name() {
         return last_name;
      }

      public String getEmail() {
         return email;
      }

      @Override
      public String toString() {
         return "Contact [id=" + id + ", first_name=" + first_name + ", last_name=" + last_name + ", email=" + email + "]";
      }
   }

<dependency>
    <groupId>org.mariadb</groupId>
    <artifactId>r2dbc-mariadb</artifactId>
</dependency>

<dependency>
    <groupId>org.mariadb</groupId>
    <artifactId>r2dbc-mariadb-0.9.1-spec</artifactId>
    <version>1.4.1</version>
</dependency>

$ mvn package

<dependency>
   <groupId>org.mariadb</groupId>
   <artifactId>r2dbc-mariadb</artifactId>
   <version>1.4.1</version>
</dependency>

<dependency>
    <groupId>io.r2dbc</groupId>
    <artifactId>r2dbc-pool</artifactId>
    <version>1.0.2.RELEASE</version>
</dependency>

$ mvn package

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Statement;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import reactor.core.publisher.Flux;

public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   public static void main(String[] argv) {

      // Initialize Connection
      initConnection();

      // Print contacts
      printContactList();

      // Close Connection
      conn.close();
   }

   public static void initConnection() {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Connection
         conn = connFactory.create().block();
      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      }
   }

   public static void printContactList() {

      try {
         // Initialize Statement
         Statement stmnt = conn.createStatement(
            "SELECT first_name, last_name, email FROM test.contacts");

         // Execute Statement and Iterate over the Result-set
         for (String contact_entry : Flux.from(stmnt.execute()).flatMap( res ->
            res.map( (row, metadata) -> {

            return String.format( "- %s %s <%s>",
               // Get First Name
               row.get(0, String.class),

               //  Get Last Name
               row.get(1, String.class),

               //Get Email
               row.get(2, String.class));
         })).toIterable()) {

            System.out.println(contact_entry);
         }
      }
      // Catch Exception
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Encountered issue querying contact list");
         e.printStackTrace();
      }
   }
}

* John Smith <john.smith@example.com>
* Jon Smith <jon.smith@example.com>

// Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;

import reactor.test.StepVerifier;

// Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Create a Database Table

         client.execute("CREATE TABLE IF NOT EXISTS test.contact" + "(id INT PRIMARY KEY AUTO_INCREMENT,"
               + "first_name VARCHAR(25)," + "last_name VARCHAR(25)," + "email VARCHAR(25)) ENGINE=InnoDB")
               .fetch()
               .rowsUpdated()
               .as(StepVerifier::create)
               .expectNextCount(1)
               .verifyComplete();

      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      } finally {

      }
   }
}

SHOW TABLES;

+----------------+
| Tables_in_test |
+----------------+
| contact        |
+----------------+

DESCRIBE contact;

+------------+-------------+------+-----+--------------------------+
| Field      | Type        | Null | Key | Default  | Extra         |
+------------+-------------+------+-----+----------+---------------+
| id         | int(11)     | NO   | PRI | NULL     | auto_increment|
| first_Name | varchar(25) | YES  |     | NULL     |               |
| last_Name  | varchar(25) | YES  |     | NULL     |               |
| email      | varchar(25) | YES  |     | NULL     |               |
+------------+-------------+------+-----+----------+---------------+

//Module Imports
import java.util.Arrays;
import java.util.List;
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;
import reactor.test.StepVerifier;

// Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Create a list or batch of SQL statements
         List<String> batch = Arrays.asList(
               "CREATE DATABASE IF NOT EXISTS test;",
               "CREATE TABLE IF NOT EXISTS test.contact_copy(id INT PRIMARY KEY AUTO_INCREMENT,first_name VARCHAR(50),last_name VARCHAR(50),email VARCHAR(250)) ENGINE=InnoDB;",
               "INSERT INTO test.contact_copy SELECT * FROM test.contact;");

         //Run the batch of SQL statements
         batch.forEach(stmt -> client.execute(stmt)
               .fetch()
               .rowsUpdated()
               .as(StepVerifier::create)
               .expectNextCount(1)
               .verifyComplete());

      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      } finally {

      }
   }
}

USE test;
SHOW TABLES;

+----------------+
| Tables_in_test |
+----------------+
| contact        |
| contact_copy   |
+----------------+

DESCRIBE contact_copy;

+------------+-------------+------+-----+--------------------------+
| Field      | Type        | Null | Key | Default  | Extra         |
+------------+-------------+------+-----+----------+---------------+
| id         | int(11)     | NO   | PRI | NULL     | auto_increment|
| first_Name | varchar(25) | YES  |     | NULL     |               |
| last_Name  | varchar(25) | YES  |     | NULL     |               |
| email      | varchar(25) | YES  |     | NULL     |               |
+------------+-------------+------+-----+----------+---------------+

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.connectionfactory.TransactionAwareConnectionFactoryProxy;
import org.springframework.data.r2dbc.core.DatabaseClient;
import reactor.test.StepVerifier;

// Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         TransactionAwareConnectionFactoryProxy proxy = new TransactionAwareConnectionFactoryProxy(connFactory);

         // Instantiate a Client
         client = DatabaseClient.create(proxy);

         // Update Data
         Contact contact = new Contact(1, "Kai", "Devi", "kai.devi@example.com");

         client.update().table(Contact.class).using(contact).then().as(StepVerifier::create).verifyComplete();

         client.execute("UPDATE test.contact SET email = 'lee.wang@example.com' WHERE id = 2").fetch().rowsUpdated()
         .then().as(StepVerifier::create).expectComplete().verify();

      } catch (IllegalArgumentException e) {
         // ...
      } finally {
         // ...
      }
   }
}

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;
import reactor.core.publisher.Mono;

//Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   // Main Process
   public static void main(String[] args) {
      // Initialize Connection Factory
      initConnectionFactory();

      // Alter Contacts Table
      alterContactsTable();
   }

   public static void initConnectionFactory() {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);
      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while creating  connection factory");
         e.printStackTrace();
      }
   }

   public static void alterContactsTable() {
      try {
         //Initialize a Connection
         conn = connFactory.create().block();

         //Create and run a Statement
         Mono.from(
            conn.createStatement("ALTER TABLE test.contact CHANGE COLUMN first_name f_name VARCHAR(25)").execute()
         ).subscribe();
      }
      // Catch Exception
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue altering contact table");
         e.printStackTrace();
      } finally {
         // Close Connection
         conn.close();
      }
   }
}

DESC test.contact;

+-----------+-------------+------+-----+---------+----------------+
| Field     | Type        | Null | Key | Default | Extra          |
+-----------+-------------+------+-----+---------+----------------+
| id        | int(11)     | NO   | PRI | NULL    | auto_increment |
| f_name    | varchar(25) | YES  |     | NULL    |                |
| last_name | varchar(25) | YES  |     | NULL    |                |
| email     | varchar(25) | YES  |     | NULL    |                |
+-----------+-------------+------+-----+---------+----------------+

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;
import reactor.core.publisher.Mono;

//Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   // Main Process
   public static void main(String[] args) {
      //Initialize Connection Factory
      initConnectionFactory();

      // Truncate table
      truncateTable();
   }

   public static void initConnectionFactory() {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);
      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      } finally {
         // ...
      }
   }

   public static void truncateTable() {
      try {

         //Initialize a Connection
         conn = connFactory.create().block();

         // Initialize and run a Statement

         Mono.from(
            conn.createStatement("TRUNCATE test.contact").execute()
         ).subscribe();
      }
      // Catch Exception
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue truncating contact table");
         e.printStackTrace();
      } finally {
         // Close Connection
         conn.close();
      }
   }
}

SELECT * FROM test.contact;

Empty set (0.000 sec)

// Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;

import reactor.test.StepVerifier;

// Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Create a Database Table

         client.execute("CREATE OR REPLACE TABLE test.contact" + "(id INT PRIMARY KEY AUTO_INCREMENT,"
               + "first_name VARCHAR(25)," + "last_name VARCHAR(25)," + "email VARCHAR(25)) ENGINE=InnoDB")
               .fetch()
               .rowsUpdated()
               .as(StepVerifier::create)
               .expectNextCount(1)
               .verifyComplete();

      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      } finally {

      }
   }
}

SHOW TABLES;

+----------------+
| Tables_in_test |
+----------------+
| contact        |
+----------------+

DESC contact;

+------------+-------------+------+-----+--------------------------+
| Field      | Type        | Null | Key | Default  | Extra         |
+------------+-------------+------+-----+----------+---------------+
| id         | int(11)     | NO   | PRI | NULL     | auto_increment|
| first_Name | varchar(25) | YES  |     | NULL     |               |
| last_Name  | varchar(25) | YES  |     | NULL     |               |
| email      | varchar(25) | YES  |     | NULL     |               |
+------------+-------------+------+-----+----------+---------------+

// Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;

import reactor.test.StepVerifier;

// Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {
      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Truncate Database Table

         client.execute("TRUNCATE TABLE test.contact").fetch()
               .rowsUpdated().as(StepVerifier::create).expectNextCount(1).verifyComplete();
      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      } finally {
      }
   }

}

// Module Imports
import java.time.Duration;
import io.r2dbc.pool.ConnectionPool;
import io.r2dbc.pool.ConnectionPoolConfiguration;
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;

public class App {

   // Attributes
   private MariadbConnectionFactory connFactory;
   private ConnectionPool pool;

   // Main Process
   public static void main(String[] args) {

      App  app = new App();
      app.createConnectionPool();
   }

   public void createConnectionPool() {

      try {
         // Configure and Create Connection Factory
         MariadbConnectionConfiguration factoryConfig = MariadbConnectionConfiguration
            .builder().host("192.0.2.1").port(3306)
            .username("db_user").password("db_user_password")
            .database("test").build();

         this.connFactory = new MariadbConnectionFactory(factoryConfig);

         // Configure Connection Pool
         ConnectionPoolConfiguration poolConfig = ConnectionPoolConfiguration
            .builder(connFactory)
            .maxIdleTime(Duration.ofMillis(1000))
            .maxSize(20)
            .build();

         this.pool = new ConnectionPool(poolConfig);
      }
      // Catch Exception
       catch (IllegalArgumentException e) {
         System.err.println("Issue creating connection pool");
         e.printStackTrace();
      } finally {
         // Close Connection
         pool.close();
      }
   }
}

// Module Imports
import java.time.Duration;
import io.r2dbc.pool.ConnectionPool;
import io.r2dbc.pool.ConnectionPoolConfiguration;
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;

public class App {

   // Attributes
   private MariadbConnectionFactory connFactory;
   private static ConnectionPool pool;
   private Connection conn;

   // Main Process
   public static void main(String[] args) {

     App app = new App();
      app.createConnectionPool();
      app.setConnection(pool);
   }

   public void createConnectionPool() {

      try {
         // Configure and Create Connection Factory
         MariadbConnectionConfiguration factoryConfig = MariadbConnectionConfiguration
            .builder().host("192.0.2.1").port(3306)
            .username("db_user").password("db_user_password")
            .database("test").build();

         this.connFactory = new MariadbConnectionFactory(factoryConfig);

         // Configure Connection Pool
         ConnectionPoolConfiguration poolConfig = ConnectionPoolConfiguration
            .builder(connFactory)
            .maxIdleTime(Duration.ofMillis(1000))
            .maxSize(20)
            .build();

         this.pool = new ConnectionPool(poolConfig);
      }
      // Catch Exception
       catch (IllegalArgumentException e) {
         System.err.println("Issue creating connection pool");
         e.printStackTrace();
      }
   }

   public void setConnection(ConnectionPool pool) {

      // Set Connection
      this.conn = pool.create().block();
      //Use Connection
      //...
   }
}

// Return Connection to the Pool
conn.close();

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Batch;
import io.r2dbc.spi.Connection;
import reactor.core.publisher.Flux;

//Main Application Class
public class App {
     // Connection Configuration
     private static MariadbConnectionConfiguration conf;
     private static MariadbConnectionFactory connFactory;
     private static Connection conn;

     // Main Process
     public static void main(String[] args) {
         //Initialize Connection Factory
         initConnectionFactory();

         // Add and Select Contacts in a batch operation
         addAndSelectContacts();
     }

     public static void initConnectionFactory() {
         try {
             // Configure the Connection
             conf = MariadbConnectionConfiguration.builder()
                  .host("192.0.2.1").port(3306)
                  .username("db_user").password("db_user_password")
                  .database("test").build();
             // Instantiate a Connection Factory
             connFactory = new MariadbConnectionFactory(conf);

         } catch (java.lang.IllegalArgumentException e) {
             System.err.println("Issue encountered while getting connection");
             e.printStackTrace();
         }
     }

     public static void addAndSelectContacts() {
         try {
             //Initialize a Connection
             conn = connFactory.create().block();
             //Create a Batch Object
             Batch batch = conn.createBatch();
             batch = batch.add("INSERT INTO test.contact (" + "first_name, last_name, email)  VALUES" + "('Kai', 'Devi', 'kai.devi@example.com'), "
                          + "('Lee', 'Wang', 'lee.wang@example.com'), " + "('Dani', 'Smith', 'dani.smith@example.com')").add("SELECT * FROM test.contact");

             for (String contact_entry : Flux.from(batch.execute()).flatMap( res -> res.map( (row, metadata) -> {
                     return String.format("- %s %s <%s>",
                         row.get(1, String.class), // Get First Name
                         row.get(2, String.class), // Get Last Name
                         row.get(3, String.class)); // Get Email
                     })).toIterable()) {
                 System.out.println(contact_entry);
             }
         }
         // Catch Exception
         catch (IllegalArgumentException e) {
             System.err.println("Issue running operation");
             e.printStackTrace();
         } finally {// Close Connection
             conn.close();
         }
     }
 }

- Kai Devi <kai.devi@example.com>
- Lee Wang <lee.wang@example.com>
- Dani Smith <dani.smith@example.com>

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Batch;
import io.r2dbc.spi.Connection;
import reactor.core.publisher.Mono;

//Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

  // Main Process
   public static void main(String[] args) {

      //Initialize Connection Factory
      initConnectionFactory();

      // Initialize the schema
      initializeSchema();

   }

   public static void initConnectionFactory() {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      }
   }

   public static void initializeSchema() {

      try {
         //Initialize a Connection
         conn = connFactory.create().block();

         //Create a Batch Object
         Batch batch = conn.createBatch();
         batch = batch.add("CREATE DATABASE IF NOT EXISTS test")
               .add("CREATE TABLE IF NOT EXISTS " +
                    "test.contact_copy (" +
                    "id INT PRIMARY KEY AUTO_INCREMENT," +
                     "first_name VARCHAR(50)," +
                     "last_name VARCHAR(50)," +
                     "email VARCHAR(250)" +
                     ") ENGINE=InnoDB")
               .add("INSERT INTO test.contact_copy SELECT * FROM test.contact");
         Mono.from(batch.execute()).subscribe();
      }
      // Catch Exception
      catch (IllegalArgumentException e) {
         System.err.println("Issue running operation");
         e.printStackTrace();
      }finally { // Close Connection
          conn.close();
      }
   }
}

SELECT TABLE_SCHEMA, TABLE_NAME, ENGINE
   FROM information_schema.TABLES
   WHERE TABLE_SCHEMA='test'
   AND TABLE_NAME='contact_copy';

+--------------+--------------+--------+
| TABLE_SCHEMA |  TABLE_NAME  | ENGINE |
+--------------+--------------+--------+
| test         | contact_copy | InnoDB |
+--------------+--------------+--------+

SELECT * FROM contact_copy;

+----+------------+-----------+------------------------+
| id | first_name | last_name | email                  |
+----+------------+-----------+------------------------+
| 1  | Kai        | Devi      | kai.devi@example.com   |
| 2  | Lee        | Wang      | lee.wang@example.com   |
| 3  | Dani       | Smith     | dani.smith@example.com |
+----+------------+-----------+------------------------+

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Statement;
import reactor.core.publisher.Mono;

// Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   // Main Process
   public static void main(String[] args) {
      //Initialize Connection Factory
      initConnectionFactory();

      // Insert a contact
      insertContact("John", "Smith", "js@example.com");
   }

   public static void initConnectionFactory() {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
            .host("192.0.2.1").port(3306)
            .username("db_user").password("db_user_password")
            .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      }
   }

   public static void insertContact(String first_name, String last_name, String email) {

      try {
         //Initialize a Connection
         conn = connFactory.create().block();

         // Initialize Statement
         Statement stmnt = conn.createStatement(
            "INSERT INTO test.contact (first_name, last_name, email) VALUES (?, ?, ?)");

         // Bind Values to Statement
         stmnt.bind(0, first_name);
         stmnt.bind(1, last_name);
         stmnt.bind(2, email);

         // Execute Statement
         Mono.from(stmnt.execute()).subscribe();

      }
      // Catch Exception
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while adding contact");
         e.printStackTrace();
      } finally {
         // Close Connection
         conn.close();
      }
   }
}

SELECT * FROM test.contact;

+----+------------+-----------+----------------+
| id | first_name | last_name | email          |
+----+------------+-----------+----------------+
|  1 | John       | Smith     | js@example.com |
+----+------------+-----------+----------------+

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Statement;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import reactor.core.publisher.Flux;

public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;
   private static Connection conn;

   public static void main(String[] argv) {

      //Initialize Connection Factory
      initConnectionFactory();

      // Print contacts
      printContactList();
   }

   public static void initConnectionFactory() {

      try {
         // Configure the Connection
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

      }
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Issue encountered while getting connection");
         e.printStackTrace();
      } finally {
      }
   }

   public static void printContactList() {

      try {
         //Initialize a Connection
         conn = connFactory.create().block();

         // Initialize Statement
         Statement stmnt = conn.createStatement(
            "SELECT first_name, last_name, email FROM test.contact");

         // Execute Statement and Iterate over the Result-set
         for (String contact_entry : Flux.from(stmnt.execute()).flatMap( res ->
            res.map( (row, metadata) -> {

            return String.format( "- %s %s <%s>",
               // Get First Name
               row.get(0, String.class),

               //  Get Last Name
               row.get(1, String.class),

               //Get Email
               row.get(2, String.class));
         })).toIterable()) {

            System.out.println(contact_entry);
         }
      }
      // Catch Exception
      catch (java.lang.IllegalArgumentException e) {
         System.err.println("Encountered issue querying contact list");
         e.printStackTrace();
      } finally {
         // Close Connection
         conn.close();
      }
   }
}

John Smith <js@example.com>

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;

import reactor.test.StepVerifier;

// Main Application Class
public class App {

   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;

   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {

      try {
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Add First Contact
         client.insert()
            .into(Contact.class)
            .using(new Contact(1, "Kai", "Devi", "kai.devi@example.com"))
            .then()
            .as(StepVerifier::create)
            .verifyComplete();

         // Add Second Contact
         client.insert()
            .into(Contact.class)
            .using(new Contact(2, "Lee", "Wang", "kai.devi@example.com"))
            .then()
            .as(StepVerifier::create)
            .verifyComplete();

         // Add Third Contact
         client.insert()
            .into(Contact.class)
            .using(new Contact(3, "Dani", "Smith", "dani.smith@example.com"))
            .then()
            .as(StepVerifier::create)
            .verifyComplete();

      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      }
   }
}

SELECT * FROM test.contact;

+----+------------+-----------+------------------------+
| id | first_name | last_name | email                  |
+----+------------+-----------+------------------------+
|  1 | John       | Smith     | john.smith@example.com |
+----+------------+-----------+------------------------+

//Module Imports
import org.mariadb.r2dbc.MariadbConnectionConfiguration;
import org.mariadb.r2dbc.MariadbConnectionFactory;
import org.springframework.data.r2dbc.core.DatabaseClient;
import reactor.test.StepVerifier;

// Main Application Class
public class App {
   // Connection Configuration
   private static MariadbConnectionConfiguration conf;
   private static MariadbConnectionFactory connFactory;

   private static DatabaseClient client;

   // Main Process
   public static void main(String[] args) {
      try {
         conf = MariadbConnectionConfiguration.builder()
              .host("192.0.2.1").port(3306)
              .username("db_user").password("db_user_password")
              .database("test").build();

         // Instantiate a Connection Factory
         connFactory = new MariadbConnectionFactory(conf);

         // Instantiate a Database Client
         client = DatabaseClient.create(connFactory);

         // Select all rows
         client.select()
            .from(Contact.class)
            .fetch().all()
            .doOnNext(it -> System.out.println(it))
            .as(StepVerifier::create)
            .expectNextCount(3)
            .verifyComplete();

         // Select the first row
         client.select()
            .from(Contact.class)
            .fetch().first()
            .doOnNext(it -> System.out.println(it))
            .as(StepVerifier::create)
            .expectNextCount(1)
            .verifyComplete();

         // Select all rows with explicit query
         client.execute("SELECT id, first_name,last_name,email FROM contact")
            .as(Contact.class)
            .filter(s -> s.fetchSize(25))
            .fetch().all()
            .doOnNext(it -> System.out.println(it))
            .as(StepVerifier::create)
            .expectNextCount(3)
            .verifyComplete();

         // Select single column
         client.select()
            .from(Contact.class)
            .map((row, rowMetadata) -> row.get("first_name", String.class))
            .all()
            .doOnNext(it -> System.out.println(it))
            .as(StepVerifier::create)
            .expectNextCount(3)
            .verifyComplete();

      } catch (IllegalArgumentException e) {
         e.printStackTrace();
      } catch (io.r2dbc.spi.R2dbcNonTransientResourceException e) {
         e.printStackTrace();
      }
   }
}

//Output from first query
Contact [id=1, first_name=Kai, last_name=Devi, email=kai.devi@example.com]
Contact [id=2, first_name=Lee, last_name=Wang, email=lee.wang@example.com]
Contact [id=3, first_name=Dani, last_name=Smith, email=dani.smith@example.com]

//Output from second query
Contact [id=1, first_name=Kai, last_name=Devi, email=kai.devi@example.com]

//Output from third query
Contact [id=1, first_name=Kai, last_name=Devi, email=kai.devi@example.com]
Contact [id=2, first_name=Lee, last_name=Wang, email=lee.wang@example.com]
Contact [id=3, first_name=Dani, last_name=Smith, email=dani.smith@example.com]

//Output from fourth query
Kai
Lee
Dani

package springdata;

// Imports the @Id annotation type, which demarcates an identifier.
//Module Imports
import org.springframework.data.annotation.Id;

// This is an Entity class
// It has the same name as the text.contact table
public class Contact {

   // The class members correspond to columns
   // in the test.contact table
   private int id;
   private String first_name;
   private String last_name;
   private String email;

   // Constructor
   public Contact() {
   }

   // Constructor
   public Contact(String first_name, String last_name, String email) {

      this.first_name = first_name;
      this.last_name = last_name;
      this.email = email;
   }

   // Constructor
   public Contact(int id, String first_name, String last_name, String email) {
      this.id = id;
      this.first_name = first_name;
      this.last_name = last_name;
      this.email = email;
   }

   // The @Id annotation indicates that this field
   // is the primary key column
   @Id
   public int getId() {
      return id;
   }

   public String getFirst_name() {
      return first_name;
   }

   public String getLast_name() {
      return last_name;
   }

   public String getEmail() {
      return email;
   }

   @Override
   public String toString() {
      return "Contact [id=" + id + ", first_name=" + first_name + ", last_name=" + last_name + ", email=" + email
            + "]";
   }
}

package springdata;

//Module Imports
import reactor.core.publisher.Flux;
import org.springframework.data.r2dbc.repository.Query;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;

// ReactiveCrudRepository<Contact, Integer>:
//   Entity Class: Contact
//   Data type of identifier: Integer
interface ContactRepository extends ReactiveCrudRepository<Contact, Integer> {

   // The Query annotation provides an SQL statement corresponding to the method
   @Query("select id, first_name, last_name, email from contact c where c.first_name = :first_name")
   Flux<Contact> findByFirstname(String firstname);

   @Query("select id, first_name, last_name, email from contact c where c.id = :id")
   Flux<Contact> findById(int id);
}

package springdata;

//Module Imports
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;

import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;

@Configuration
@EnableR2dbcRepositories(basePackageClasses = ContactRepository.class)
@ComponentScan(basePackageClasses = RepositoryService.class)
class ApplicationConfig extends AbstractR2dbcConfiguration {
   @Bean
   public ConnectionFactory connectionFactory() {
      return ConnectionFactories.get("r2dbc:mariadb://connr2dbc_test:db_user_password@192.0.2.50:3306/test");
   }
}

package springdata;

//Module Imports
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.stereotype.Service;

import reactor.test.StepVerifier;

// The @Service annotation indicates that the class is a "Service".
// Spring Data Repositories framework auto-discovers the service class
// through classpath scanning because we have set the @ComponentScan
// annotation in ApplicationConfig to scan for RepositoryService.class.
@Service
public class RepositoryService {

   // The @Autowired annotation indicates that the field is to be autowired
   // by Spring's dependency injection facilities.
   @Autowired
   private static ContactRepository repository;

   // The ApplicationContext provides Bean factory methods for
   // accessing application components.
   private static ApplicationContext ctx;

   public static void main(String[] args) {

      try {
         // The AnnotationConfigApplicationContext class is a standalone application context,
         // accepting component classes as input, in particular @Configuration-annotated
         // classes such as the ApplicationConfig class we developed.
         ctx = new AnnotationConfigApplicationContext(ApplicationConfig.class);

         // Returns the bean instance that uniquely matches the ContactRepository.class
         repository = ctx.getBean(ContactRepository.class);
         RepositoryService repoService = new RepositoryService();
         repoService.crud();
      } catch (Exception e) {
         System.out.println();
      } finally {
         // ...
      }
   }

   public void crud() {

      // Print number of rows
      System.out.println("Number of contacts in database is " + repository.count().block());

      // Delete all data
      repository.deleteAll().block();

      // Print number of rows again
      System.out.println("Number of contacts in database is " + repository.count().block());

      // Insert one row
      // ID is auto-generated
      Contact contact = new Contact("John", "Smith", "john.smith@gmail.com");
      repository.save(contact)
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

      // Insert another row
      // ID is auto-generated
      contact = new Contact("Johnny", "Smith", "johnny.smith@gmail.com");
      repository.save(contact)
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

      // Insert another row
      // ID is auto-generated
      contact = new Contact("Joe", "Smith", "joe.smith@gmail.com");
      repository.save(contact)
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

      // Print all rows
      repository.findAll()
         .doOnNext(it -> System.out.println(it)).as(StepVerifier::create)
         .expectNextCount(3)
         .verifyComplete();

      // Print rows with first name "John"
      repository.findByFirstname("John")
         .doOnNext(it -> System.out.println(it))
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

      // Print row with ID 1
      repository.findById(1)
         .doOnNext(it -> System.out.println(it))
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

       // Update email for contact with id 1
      // ID is explicitly provided
      contact = new Contact(1, "John", "Smith", "johnsmith@gmail.com");
      repository.save(contact)
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();

      // Print rows with first name "John"
      repository.findByFirstname("John")
         .doOnNext(it -> System.out.println(it))
         .as(StepVerifier::create)
         .expectNextCount(1)
         .verifyComplete();
   }
}

Number of contacts in database is 0
Number of contacts in database is 0
Contact [id=1, first_name=John, last_name=Smith, email=john.smith@example.com]
Contact [id=2, first_name=Johnny, last_name=Smith, email=johnny.smith@example.com]
Contact [id=3, first_name=Joe, last_name=Smith, email=joe.smith@example.com]
Contact [id=1, first_name=John, last_name=Smith, email=john.smith@example.com]
Contact [id=1, first_name=John, last_name=Smith, email=john.smith@example.com]
Contact [id=1, first_name=John, last_name=Smith, email=johnsmith@example.com]

SELECT * FROM test.contact;

+----+------------+-----------+---------------------------+
| id | first_name | last_name | email                     |
+----+------------+-----------+---------------------------+
|  1 | John       | Smith     | johnsmith@example.com     |
+----+------------+-----------+---------------------------+
|  2 | Johnny     | Smith     | johnny.smith@example.com  |
+----+------------+-----------+---------------------------+
|  3 | Joe        | Smith     | joe.smith@example.com     |
+----+------------+-----------+---------------------------+