Hologres:Consume Binlog via JDBC

Publication

A publication defines which table's changes are made available for logical replication. In Hologres, a publication is bound to exactly one physical table, and that table must have Binlog enabled.

CREATE PUBLICATION <name> FOR TABLE <table_name>;

CREATE PUBLICATION hg_publication_test_1 FOR TABLE test_message_src;

SELECT * FROM pg_publication;

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

SELECT * FROM pg_publication_tables;

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

DROP PUBLICATION <name>;

DROP PUBLICATION hg_publication_test_1;

Replication slot

A replication slot tracks the consumption progress for a publication and enables resumable transmission. After a failover, a consumer can recover from the last committed checkpoint recorded in the slot.

Only superusers and users with the Replication Role can create and use replication slots. To grant or revoke the Replication Role:

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

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

user_name is an Alibaba Cloud account ID or a Resource Access Management (RAM) user. See Account overview.

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

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

SELECT * FROM hologres.hg_replication_slot_properties;

        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)

SELECT hg_get_logical_replication_slot_parallelism('hg_replication_slot_1');

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

SELECT * FROM hologres.hg_replication_progress;

        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)

CALL hg_drop_logical_replication_slot('<replication_slot_name>');

CALL hg_drop_logical_replication_slot('hg_replication_slot_1');

Step 1: Add dependencies

Use JDBC 42.2.18 or later. Add the following dependencies to your pom.xml:

<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>

Step 2: Connect and consume

The following example connects to a replication slot, reads Binlog records from one shard, and prints each record.

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();
            }
        }
    }
}

Parameters

`withSlotName` behavior by version:

`withSlotOption` parameters:

Default start behavior (when neither start_lsn nor start_time is set):

BinlogRecord fields

BinlogRecord exposes the following Binlog system fields:

For a full field reference, see Subscribe to Hologres Binlog data.

Commit the checkpoint

After consuming Binlog data, commit the checkpoint by calling the commit LSN function. This serves two purposes: it lets the server know which WAL segments are safe to archive or discard, and it provides a recovery point so consumption can resume from the correct position after a failover.

The sample code above does not include this step—add it based on your application's recovery requirements.

Step 1: Add dependencies

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

Step 2: Connect and consume

The following example subscribes to a table, processes each record, saves a per-shard checkpoint, and retries on failure.

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();
            }
        }
    }
}

Parameters

`Subscribe` types:

Version note: Before V2.1, both tableName and slotName are required. Starting from V2.1, only tableName is required (equivalent to using the fixed slot name hg_table_name_slot).

`HoloConfig` Binlog parameters:

The hologres.hg_replication_progress table is missing or empty

If the table doesn't exist or shows no data after you commit consumption progress, check the following:

Consuming by table name without a replication slot. If you didn't set withSlotName (JDBC) or didn't specify a slot (Holo-Client), progress tracking is not supported. The table will not be created or updated. Switch to replication slot-based consumption to enable progress tracking.

First-time consumption on a read-only secondary instance. On secondary instances running Hologres earlier than V2.0.18, the hologres.hg_replication_progress table cannot be created during the first Binlog consumption. Consume from the primary instance once, then switch back to the secondary instance.

If neither applies, contact support through the Hologres DingTalk group. See How do I get more online support?.