Hologres:Consumo de Binlog mediante JDBC

Crear una publicación

CREATE PUBLICATION <name> FOR TABLE <table_name>;

Ejemplo:

CREATE PUBLICATION hg_publication_test_1 FOR TABLE test_message_src;

Consultar publicaciones

SELECT * FROM pg_publication;

Salida de ejemplo:

         pubname        | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
------------------------+----------+--------------+-----------+-----------+-----------+-------------
 hg_publication_test_1  |    16728 | f            | t         | t         | t         | t
(1 row)

Para obtener más información sobre los tipos de eventos de Binlog, consulte Binlog format and principles.

Consultar tablas en una publicación

SELECT * FROM pg_publication_tables;

Salida de ejemplo:

         pubname        | schemaname |    tablename
------------------------+------------+------------------
 hg_publication_test_1  | public     | test_message_src
(1 row)

Eliminar una publicación

DROP PUBLICATION <name>;

Ejemplo:

DROP PUBLICATION hg_publication_test_1;

-- Grant the Replication Role to a user.
ALTER ROLE <user_name> REPLICATION;

-- Revoke the Replication Role from a user.
ALTER ROLE <user_name> NOREPLICATION;

Crear un slot de replicación

CALL hg_create_logical_replication_slot('<replication_slot_name>', 'hgoutput', '<publication_name>');

Ejemplo:

CALL hg_create_logical_replication_slot('hg_replication_slot_1', 'hgoutput', 'hg_publication_test_1');

Consultar slots de replicación

SELECT * FROM hologres.hg_replication_slot_properties;

Salida de ejemplo:

        slot_name       | property_key |      property_value
------------------------+--------------+------------------------
 hg_replication_slot_1  | plugin       | hgoutput
 hg_replication_slot_1  | publication  | hg_publication_test_1
 hg_replication_slot_1  | parallelism  | 1
(3 rows)

Consultar el paralelismo requerido

Dado que Hologres es una base de datos distribuida, los datos de cada tabla se distribuyen en varios shards. Se requiere una conexión por shard para consumir la tabla completa. Para comprobar cuántas conexiones concurrentes requiere hg_replication_slot_1:

SELECT hg_get_logical_replication_slot_parallelism('hg_replication_slot_1');

Salida de ejemplo:

 hg_get_logical_replication_slot_parallelism
---------------------------------------------
                                          20

Consultar el progreso del consumo

La tabla hologres.hg_replication_progress registra el desplazamiento del consumidor que usted confirma explícitamente.

SELECT * FROM hologres.hg_replication_progress;

Salida de ejemplo:

        slot_name       | parallel_index | lsn
------------------------+----------------+-----
 hg_replication_slot_1  |              0 |  66
 hg_replication_slot_1  |              1 | 122
 hg_replication_slot_1  |              2 | 119
(3 rows)

Eliminar un slot de replicación

Elimine un slot cuando ya no lo necesite para evitar la acumulación de WAL. Los slots no se cierran automáticamente cuando se detiene un trabajo de consumo; debe eliminarlos manualmente.

CALL hg_drop_logical_replication_slot('<replication_slot_name>');

Ejemplo:

CALL hg_drop_logical_replication_slot('hg_replication_slot_1');

<dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.3.8</version>
</dependency>
<!-- Used to get the table schema and decode Binlog records -->
<dependency>
    <groupId>com.alibaba.hologres</groupId>
    <artifactId>holo-client</artifactId>
    <version>2.2.10</version>
</dependency>

import com.alibaba.hologres.client.HoloClient;
import com.alibaba.hologres.client.HoloConfig;
import com.alibaba.hologres.client.impl.binlog.HoloBinlogDecoder;
import com.alibaba.hologres.client.model.Record;
import com.alibaba.hologres.client.model.TableSchema;

import org.postgresql.PGConnection;
import org.postgresql.PGProperty;
import org.postgresql.replication.LogSequenceNumber;
import org.postgresql.replication.PGReplicationStream;

import java.nio.ByteBuffer;
import java.sql.Connection;
import java.sql.DriverManager;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class Test {

    public static void main(String[] args) throws Exception {

        String username = "";
        String password = "";
        String url = "jdbc:postgresql://Endpoint:Port/db_test";

        // Set up the JDBC connection for replication.
        Properties properties = new Properties();
        PGProperty.USER.set(properties, username);
        PGProperty.PASSWORD.set(properties, password);
        PGProperty.ASSUME_MIN_SERVER_VERSION.set(properties, "9.4");
        // Required for Binlog consumption.
        PGProperty.REPLICATION.set(properties, "database");

        try (Connection connection = DriverManager.getConnection(url, properties)) {

            // Each PGReplicationStream consumes one shard. Set shardId for each stream.
            int shardId = 0;
            PGConnection pgConnection = connection.unwrap(PGConnection.class);
            PGReplicationStream pgReplicationStream = pgConnection.getReplicationAPI()
                    .replicationStream()
                    .logical()
                    // V2.1+: two options
                    // Option 1: specify the replication slot name. table_name is ignored.
                    // Option 2: omit withSlotName and set table_name in withSlotOption instead.
                    .withSlotName("slot_name")
                    .withSlotOption("table_name", "public.test_message_src")
                    .withSlotOption("parallel_index", shardId)
                    .withSlotOption("batch_size", "1024")
                    .withSlotOption("start_time", "2021-01-01 00:00:00")
                    .withSlotOption("start_lsn", "0")
                    .start();

            // HoloClient is needed to get the table schema for decoding.
            HoloConfig holoConfig = new HoloConfig();
            holoConfig.setJdbcUrl(url);
            holoConfig.setUsername(username);
            holoConfig.setPassword(password);
            HoloClient client = new HoloClient(holoConfig);

            // Decode raw binary Binlog data into BinlogRecord objects.
            TableSchema schema = client.getTableSchema("test_message_src", true);
            HoloBinlogDecoder decoder = new HoloBinlogDecoder(schema);

            // Track the last consumed LSN for checkpoint recovery.
            Long currentLsn = 0L;

            ByteBuffer byteBuffer = pgReplicationStream.readPending();
            while (true) {
                if (byteBuffer != null) {
                    List<BinlogRecord> records = decoder.decode(shardId, byteBuffer);
                    Long latestLsn = 0L;
                    for (BinlogRecord record : records) {
                        latestLsn = record.getBinlogLsn();
                        // Process the record here.
                        System.out.println("lsn: " + latestLsn + ", record: " + Arrays.toString(record.getValues()));
                    }
                    // Save the latest LSN as the recovery point.
                    currentLsn = latestLsn;
                    pgReplicationStream.forceUpdateStatus();
                }
                byteBuffer = pgReplicationStream.readPending();
            }
        }
    }
}

<dependency>
    <groupId>com.alibaba.hologres</groupId>
    <artifactId>holo-client</artifactId>
    <version>2.2.10</version>
</dependency>

import com.alibaba.hologres.client.BinlogShardGroupReader;
import com.alibaba.hologres.client.Command;
import com.alibaba.hologres.client.HoloClient;
import com.alibaba.hologres.client.HoloConfig;
import com.alibaba.hologres.client.Subscribe;
import com.alibaba.hologres.client.exception.HoloClientException;
import com.alibaba.hologres.client.impl.binlog.BinlogOffset;
import com.alibaba.hologres.client.model.binlog.BinlogHeartBeatRecord;
import com.alibaba.hologres.client.model.binlog.BinlogRecord;

import java.util.HashMap;
import java.util.Map;

public class HoloBinlogExample {

    public static BinlogShardGroupReader reader;

    public static void main(String[] args) throws Exception {
        String username = "";
        String password = "";
        String url = "jdbc:postgresql://ip:port/database";
        String tableName = "test_message_src";
        String slotName = "hg_replication_slot_1";

        HoloConfig holoConfig = new HoloConfig();
        holoConfig.setJdbcUrl(url);
        holoConfig.setUsername(username);
        holoConfig.setPassword(password);
        holoConfig.setBinlogReadBatchSize(128);
        holoConfig.setBinlogIgnoreDelete(true);
        holoConfig.setBinlogIgnoreBeforeUpdate(true);
        holoConfig.setBinlogHeartBeatIntervalMs(5000L);
        HoloClient client = new HoloClient(holoConfig);

        // Get the shard count to initialize per-shard checkpoint tracking.
        int shardCount = Command.getShardCount(client, client.getTableSchema(tableName));

        // Initialize the checkpoint map with LSN 0 for each shard.
        Map<Integer, Long> shardIdToLsn = new HashMap<>(shardCount);
        for (int i = 0; i < shardCount; i++) {
            shardIdToLsn.put(i, 0L);
        }

        // Before V2.1: tableName and slotName are both required.
        // V2.1 and later: tableName is enough (slotName defaults to "hg_table_name_slot").
        Subscribe subscribe = Subscribe.newStartTimeBuilder(tableName, slotName)
                .setBinlogReadStartTime("2021-01-01 12:00:00")
                .build();

        reader = client.binlogSubscribe(subscribe);

        BinlogRecord record;
        int retryCount = 0;

        while (true) {
            try {
                if (reader.isCanceled()) {
                    // Re-create the reader using the last saved checkpoint.
                    reader = client.binlogSubscribe(subscribe);
                }
                while ((record = reader.getBinlogRecord()) != null) {
                    if (record instanceof BinlogHeartBeatRecord) {
                        // All data up to this heartbeat's timestamp has been consumed on this shard.
                        continue;
                    }

                    // Process the record here.
                    System.out.println(record);

                    // Save the checkpoint so consumption can resume from here on failure.
                    shardIdToLsn.put(record.getShardId(), record.getBinlogLsn());
                    retryCount = 0;
                }
            } catch (HoloClientException e) {
                if (++retryCount > 10) {
                    throw new RuntimeException(e);
                }
                System.out.println(String.format(
                        "Binlog read failed: %s. Retrying (%d/10)...", e.getMessage(), retryCount));

                // Wait before retrying, with increasing backoff.
                Thread.sleep(5000L * retryCount);

                // Resume from the saved per-shard checkpoint.
                Subscribe.OffsetBuilder subscribeBuilder = Subscribe.newOffsetBuilder(tableName, slotName);
                for (int i = 0; i < shardCount; i++) {
                    subscribeBuilder.addShardStartOffset(i,
                            new BinlogOffset().setSequence(shardIdToLsn.get(i)));
                }
                subscribe = subscribeBuilder.build();
                reader.cancel();
            }
        }
    }
}