The Wonders of Postgres Logical Decoding Messages

Do you know there’s a perform in Postgres that permits you to write knowledge which you’ll be able to’t question? A perform that permits you to persist knowledge in all types and shapes however which is able to by no means present up in any desk? Let me let you know about pg_logical_emit_message()! It’s a Postgres perform that permits you to write messages to the write-ahead log (WAL) of the database.

You’ll be able to then use logical decoding—Postgres’ change knowledge seize functionality—to retrieve these messages from the WAL, course of them, and relay them to exterior customers.

On this article, we’ll discover how you can reap the benefits of this characteristic for implementing three totally different use instances:

• Propagating knowledge between microservices by way of the outbox sample
• Utility logging
• Enriching audit logs with metadata

For retrieving logical decoding messages from Postgres we’re going to use Debezium, a well-liked open-source platform for log-based change knowledge seize (CDC), which may stream knowledge adjustments from a big number of databases into knowledge streaming platforms like Apache Kafka or AWS Kinesis.

We’ll additionally use Apache Flink and the Flink CDC mission, which seamlessly integrates Debezium into the Flink ecosystem, for enriching and routing uncooked change occasion streams. You’ll be able to be taught extra concerning the foundations of change knowledge seize and Debezium in this talk from QCon San Francisco.

Logical Decoding Messages 101

Earlier than diving into particular use instances, let’s check out how logical decoding messages might be emitted and consumed. To observe alongside, be sure that to have Docker put in in your machine. Begin by trying out this example project from GitHub:

git clone https://github.com/decodableco/examples.git
cd examples/postgres-logical-decoding

The mission accommodates a Docker Compose file for working a Postgres database, which is enabled for logical replication already. Begin it like so:

docker compose up

Then, in one other terminal window, connect with that Postgres occasion utilizing the pgcli command line consumer:

docker run --tty --rm -i 
  --network logical-decoding-network 
  quay.io/debezium/tooling:1.2 bash -c 
  'pgcli postgresql://postgresuser:postgrespw@postgres:5432/demodb'

Subsequent, it is advisable to create a replication slot. A replication slot represents one particular stream of adjustments coming from a Postgres database and retains observe of how far a client has processed this stream. For this objective, it shops the newest log sequence quantity (LSN) that the slot’s client has processed and acknowledged.

Every slot has a reputation and an assigned decoding plug-in which defines the format of that stream. Create a slot utilizing the “test_decoding” plug-in, which emits adjustments in a easy text-based protocol, like this:

postgresuser@postgres:demodb> SELECT * FROM pg_create_logical_replication_slot('demo_slot', 'test_decoding');


+-------------+-----------+
| slot_name   | lsn       |
|-------------+-----------|
| demo_slot   | 0/1A24E38 |
+-------------+-----------+

For manufacturing eventualities it is suggested to make use of the pgoutput plug-in, which emits change occasions utilizing an environment friendly Postgres-specific binary format and is accessible by default in Postgres since model 10. Different generally used choices embody the Decoderbufs plug-in (based mostly on the Google Protocol Buffers format) and wal2json (emitting change occasions as JSON).

Modifications are usually retrieved from distant purchasers similar to Debezium by establishing a replication stream with the database. Alternatively, you need to use the perform pg_logical_slot_get_changes(), which helps you to fetch adjustments from a given replication slot by way of SQL, optionally studying solely as much as a particular LSN (the primary NULL parameter) or solely a particular variety of adjustments (the second NULL parameter). This is useful for testing functions:

postgresuser@postgres:demodb> SELECT * FROM pg_logical_slot_get_changes('demo_slot', NULL, NULL);

+-------+-------+--------+
| lsn   | xid   | knowledge   |
|-------+-------+--------|
+-------+-------+--------+

No adjustments needs to be returned at this level. Let’s insert a logical decoding message utilizing the pg_logical_emit_message() perform:

postgresuser@postgres:demodb> SELECT * FROM pg_logical_emit_message(true, 'context', 'Hi there World!');

+---------------------------+
| pg_logical_emit_message   |
|---------------------------|
| 0/1A24F68                 |
+---------------------------+

The perform has three parameters:

• transactional: a boolean flag indicating whether or not the message needs to be transactional or not; when issued whereas a transaction is pending and that transaction will get rolled again finally, a transactional message wouldn’t be emitted, whereas a non-transactional message can be written to the WAL however
• prefix: a textual identifier for categorizing messages; as an example, this might point out the kind of a particular message
• content material: the precise payload of the message, both as textual content or binary knowledge; you’ve full flexibility of what to emit right here, e.g., in regard to format, schema, and semantics

Whenever you retrieve adjustments from the slot once more after having emitted a message, you now ought to see three change occasions: a BEGIN and a COMMIT occasion for the implicitly created transaction when emitting the occasion, and the “Hi there World!” message itself. Notice that this message doesn’t seem in any Postgres desk or view as can be the case when including knowledge utilizing the INSERT assertion; this message is solely current within the database’s transaction log.

There are a couple of different helpful capabilities coping with logical decoding messages and replication slots, together with the next:

• pg_logical_slot_get_binary_changes(): retrieves binary messages from a slot
• pg_logical_slot_peek_changes(): permits to check out adjustments from a slot with out advancing it
• pg_replication_slot_advance(): advances a replication slot
• pg_drop_replication_slot(): deletes a replication slot

You can also question the pg_replication_slots view for analyzing the present standing of your replication slots, newest confirmed LSN, and extra.

Use Instances

Having mentioned the foundations of logical decoding messages, let’s now discover a couple of use instances of this convenient Postgres API.

The Outbox Sample

For microservices, it’s a standard requirement that, when processing a request, a service must replace its personal database and concurrently ship a message to different companies. For example, contemplate a “achievement” service in an e-commerce state of affairs: when the standing of a cargo adjustments from READY_TO_SHIP to SHIPPED, the cargo’s report within the achievement service database must be up to date accordingly, but additionally a message needs to be despatched to the “buyer” service in order that it could possibly replace the client’s account historical past and set off an e mail notification for the client.

Now, when utilizing knowledge streaming platforms like Apache Kafka for connecting your companies, you possibly can’t reliably implement this state of affairs by simply letting the achievement service situation its native database transaction after which ship a message by way of Kafka. The reason being that it’s not supported to have shared transactions for a database and Kafka (in technical phrases, Kafka can’t take part in distributed transaction protocols like XA). Whereas every part appears to be like fantastic on the floor, you possibly can find yourself with an inconsistent state in case of failures. The database transaction might get dedicated, however sending out the notification by way of Kafka fails. Or, the opposite means round: the customer support will get notified, however the native database transaction will get rolled again.

Whereas you will discover this type of implementation in lots of purposes, all the time bear in mind: “Friends don’t let friends do dual writes”! An answer to this drawback is the outbox sample: as a substitute of attempting to replace two sources without delay (a database and Kafka), you solely replace a single one—the service’s database. When updating the cargo state within the database, you additionally write the message to be despatched to an outbox desk; this occurs as a part of one shared transaction, i.e., making use of the atomicity ensures you get from ACID transactions. Both the cargo state replace and the outbox message get continued, or none of them do. You then use change knowledge seize to retrieve any inserts from the outbox within the database and propagate them to customers.

Extra details about the outbox sample might be present in this blog post on the Debezium weblog. One other useful resource is this article on InfoQ which discusses how the outbox sample can be utilized as the inspiration for implementing Sagas between a number of companies. Within the following, I’d wish to dive into one specific implementation strategy for the sample. As an alternative of inserting outbox occasions in a devoted outbox desk, the concept is to emit them simply as logical decoding messages to the WAL.

There are professionals and cons to both strategy. What makes the route by way of logical decoding messages compelling is that it avoids any housekeeping wants. In contrast to with an outbox desk, there’s no must take away messages after they’ve been consumed from the transaction log. Additionally, this emphasizes the character of an outbox being an append-only medium: messages mustn’t ever be modified after being added to the outbox, which could occur by chance with a table-based strategy.

Relating to the content material of outbox messages, you’ve full flexibility there on the whole. Sticking to the e-commerce area from above, it might, as an example, describe a cargo serialized as JSON, Apache Avro, Google Protocol Buffers, or another format you select. What’s necessary to bear in mind is that whereas the message content material doesn’t adhere to any particular desk schema from a database perspective, it’s topic to an (ideally specific) contract between the sending software and any message customers. Specifically, the schema of any emitted occasions ought to solely be modified should you remember the impression on customers and backward compatibility.

One generally used strategy is to take a look at the design of outbox occasions and their schemas from a domain-driven design perspective. Particularly, Debezium recommends that your messages have the following attributes:

• id: a singular message id, e.g., a UUID, which customers can use for deduplication functions
• mixture sort: describes the type of mixture an occasion is about, e.g., “buyer,” “cargo,” or “buy order”; when propagating outbox occasions by way of Kafka or different streaming platforms, this can be utilized for sending occasions of 1 mixture sort to a particular subject
• mixture id: the id of the mixture an occasion is about, e.g., a buyer or order id; this can be utilized because the report key in Kafka, thus making certain all occasions pertaining to 1 mixture will go to the identical subject partition and ensuring customers obtain these occasions within the right order
• payload: the precise message payload; in contrast to “uncooked” table-level CDC occasions, this generally is a wealthy construction, representing a complete mixture and all its elements, which within the database itself might unfold throughout a number of tables

Determine 1: Routing outbox occasions from the transaction log to totally different Kafka matters

Sufficient of the speculation—let’s see how a database transaction might look, which emits a logical decoding message with an outbox occasion. Within the accompanying GitHub repository, you will discover a Docker Compose file for spinning up all of the required parts and detailed instructions for working the whole instance your self. Emit an outbox message like this:

postgresuser@postgres:demodb> SELECT * FROM pg_logical_emit_message(
  true,
  'outbox',
  '{
    "id" : "298c2cc3-71bb-4d2b-b5b4-1b14006d56e6",
    "aggregate_type" : "cargo",
    "aggregate_id" : 42,
    "payload" : {
      "customer_id" : 7398,
      "item_id" : 8123,
      "standing" : "SHIPPED",
      "numberOfPackages" : 3,
      "deal with" : "Bob Summers, 12 Most important St., 90210, Los Angeles/CA, US"
    }
  }'
 );

This creates a transactional message (i.e., it could not be emitted if the transaction aborts, e.g., due to a constraint violation of one other report inserted in the identical transaction). It makes use of the “outbox” prefix (permitting it to differentiate it from messages of different varieties) and accommodates a JSON message because the precise payload.

Relating to retrieving change occasions and propagating them to Kafka, the main points rely upon how precisely Debezium, because the underlying CDC instrument, is deployed. When used with Kafka Connect, Debezium gives a single message transform (SMT) that helps outbox tables and, as an example, routes outbox occasions to totally different matters in Kafka based mostly on a configurable column containing the mixture sort. Nevertheless, this SMT doesn’t but help utilizing logical decoding messages because the outbox format.

When utilizing Debezium by way of Flink CDC, you could possibly implement the same logic utilizing a customized KafkaRecordSerializationSchema which routes outbox occasions to the appropriate Kafka subject and propagates the mixture id to the Kafka message key, thus making certain right ordering semantics. A primary implementation of this might appear like this (you will discover the whole supply code, together with the utilization of this serializer in a Flink job here):

public class OutboxSerializer implements KafkaRecordSerializationSchema<ChangeEvent> {

  non-public static ultimate lengthy serialVersionUID = 1L;

  non-public ObjectMapper mapper;

  @Override
  public ProducerRecord<byte[], byte[]> serialize(ChangeEvent aspect, 
      KafkaSinkContext context, Lengthy timestamp) {
    strive {
      JsonNode content material = aspect.getMessage().getContent();


      ProducerRecord<byte[], byte[]> report =
          new ProducerRecord<byte[], byte[]>(
        content material.get("aggregate_type").asText(),
        content material.get("aggregate_id").asText().getBytes(Charsets.UTF_8),
        mapper.writeValueAsBytes(content material.get("payload"))
      );

      report.headers().add("message_id",      
          content material.get("id").asText().getBytes(Charsets.UTF_8));

      return report;
    }
    catch (JsonProcessingException e) {
      throw new IllegalArgumentException(
          "Could not serialize outbox message", e);
    }
  }

  @Override
  public void open(InitializationContext context,
    KafkaSinkContext sinkContext) throws Exception {


    mapper = new ObjectMapper();
    SimpleModule module = new SimpleModule();
    module.addDeserializer(Message.class, new MessageDeserializer());
        mapper.registerModule(module);
  }
}

With that Flink job in place, you’ll be capable of look at the outbox message on the “cargo” Kafka subject like so:

docker run --tty --rm 
  --network logical-decoding-network 
  quay.io/debezium/tooling:1.2 
  kcat -b kafka:9092 -C -o starting -q -t cargo 
  -f '%okay -- %h -- %sn'

42 -- message_id=298c2cc3-71bb-4d2b-b5b4-1b14006d56e6 -- {"customer_id":7398,"item_id":8123,"standing":"SHIPPED","numberOfPackages":3,"deal with":"Bob Summers, 12 Most important St., 90210, Los Angeles/CA, US"}

The subject identify corresponds to the required mixture sort, i.e., should you have been to situation outbox occasions for different mixture varieties, they’d be routed to totally different matters accordingly. The message secret is 42, matching the mixture id. The combination id is propagated as a Kafka message header, enabling customers to implement environment friendly deduplication by preserving observe of the ids they’ve already acquired and processed and ignoring any potential duplicates they could encounter. Lastly, the payload of the outbox occasion is propagated because the Kafka message worth.

Specifically, in bigger organizations with a various set of occasion producers and customers, it is smart to align on a shared occasion envelope format, which standardizes frequent attributes like occasion timestamp, origin, partitioning key, schema URLs, and others. The CloudEvents specification is useful right here, particularly for outlining occasion varieties and their schemas. It’s an choice price contemplating to have your purposes emit outbox occasions adhering to the CloudEvents commonplace.

Logging

Whereas log administration of contemporary purposes usually occurs via devoted platforms like Datadog or Splunk, which ingest adjustments from devoted APIs or logs within the file system, it generally might be handy to persist log messages within the database of an software. Log libraries such because the broadly used log4j 2 present database-backed appenders for this objective. These will usually require a second connection for the logger, although, as a result of in case of a rollback of an software transaction itself, you continue to (and particularly then) wish to write out any log messages, serving to you with failure evaluation.

Non-transactional logical decoding messages generally is a good technique of utilizing a single connection and nonetheless making certain that log messages persist, additionally when a transaction is rolled again. For instance, let’s contemplate the next state of affairs with two transactions, one in every of which is dedicated and one rolled again:

Determine 2: Utilizing non-transactional logical decoding messages for logging functions

To observe alongside, run the next sequence of statements within the pgcli shell:

–- Assuming this desk: CREATE TABLE knowledge (id INTEGER, worth TEXT);

BEGIN;
INSERT INTO knowledge(id, worth) VALUES('1', 'foo');
SELECT * FROM pg_logical_emit_message(false, 'log', 'OK');
INSERT INTO knowledge(id, worth) VALUES('2', 'bar');
COMMIT;

BEGIN;
INSERT INTO knowledge(id, worth) VALUES('3', 'baz');
SELECT * FROM pg_logical_emit_message(false, 'log', 'ERROR');
INSERT INTO knowledge(id, worth) VALUES('4', 'qux');
ROLLBACK;

The primary transaction inserts two information in a brand new desk, “knowledge” and in addition emits a logical decoding message. The second transaction applies comparable adjustments however then is rolled again. When retrieving the change occasions from the replication slot (utilizing the “testing” decoding plug-in as proven above), the next occasions will likely be returned:

See Also

We every have a mean of 100 on-line accounts. Right here’s how to verify they aren’t a nightmare for your loved ones should you die

postgresuser@postgres:demodb> SELECT * FROM pg_logical_slot_peek_changes('demo_slot', NULL, NULL) order by lsn;

+-----------+-------+------------------------------------------------------------+
| lsn       | xid   | knowledge                                                       |
|-----------+-------+------------------------------------------------------------|
| 0/1A483F8 | 768   | BEGIN 768                                                  |
| 0/1A504B8 | 768   | desk public.knowledge: INSERT: id[integer]:1 worth[text]:'foo' |
| 0/1A50530 | 768   | message: transactional: 0 prefix: log, sz: 2 content material:OK    |
| 0/1A50530 | 768   | desk public.knowledge: INSERT: id[integer]:2 worth[text]:'bar' |
| 0/1A509B8 | 768   | COMMIT 768                                                 |
| 0/1A50A38 | 769   | message: transactional: 0 prefix: log, sz: 5 content material:ERROR |
+-----------+-------+------------------------------------------------------------+

As anticipated, there are two INSERT occasions and the log message for the primary transaction. Nevertheless, there are not any change occasions for the aborted transaction for the INSERT statements, because it was rolled again. However because the logical decoding message was non-transactional, it nonetheless was written to the WAL and might be retrieved. I.e., you really can have that cake and eat it too!

Audit Logs

In enterprise purposes, preserving an audit log of your knowledge is a standard requirement, i.e., an entire path of all of the adjustments performed to a database report, similar to a purchase order order or a buyer.

There are a number of attainable approaches for constructing such an audit log; one in every of them is to repeat earlier report variations right into a separate historical past desk each time an information change is made. Arguably, this will increase software complexity. Relying on the precise implementation technique, you might need to deploy triggers for all of the tables that needs to be audited or add libraries similar to Hibernate Envers, an extension to the favored Hibernate object-relational mapping instrument. As well as, there’s a efficiency impression, because the audit information are inserted as a part of the appliance’s transactions, thus growing write latency.

Change knowledge seize is an attention-grabbing various for constructing audit logs: extracting knowledge adjustments from the database transaction log requires no adjustments to writing purposes. A change occasion stream, with occasions for all of the inserts, updates, and deletes executed for a desk—e.g., continued as a subject in Apache Kafka, whose information are immutable by definition—may very well be thought-about a easy type of an audit log. Because the CDC course of runs asynchronously, there’s no latency impression on writing transactions.

One shortcoming of this strategy—at the least in its most elementary type—is that it doesn’t seize contextual metadata, like the appliance consumer making a given change, consumer info like gadget configuration or IP deal with, use case identifiers, and many others. Usually, this knowledge is just not saved within the enterprise tables of an software and thus isn’t uncovered in uncooked change knowledge occasions. 

The mixture of logical decoding messages and stream processing, with Apache Flink, can present an answer right here. At first of every transaction, the supply software writes all of the required metadata right into a message; compared to writing a full historical past entry for every modified report, this simply provides a small overhead on the write path. You’ll be able to then use a easy Flink job for enriching all the following change occasions from that very same transaction with that metadata. As all change occasions emitted by Debezium include the id of the transaction they originate from, together with logical decoding messages, correlating the occasions of 1 transaction isn’t difficult. The next picture exhibits the overall thought:

Determine 3: Enriching knowledge change occasions with transaction-scoped audit metadata

Relating to implementing this logic with Apache Flink, you are able to do this utilizing a reasonably easy mapping perform, particularly by implementing the RichFlatMapFunction interface, which lets you mix the enrichment performance and the removing of the unique logical decoding messages in a single operator name:

public void flatMap(String worth, Collector<String> out)
    throws Exception {


  ChangeEvent changeEvent = mapper.readValue(worth, ChangeEvent.class);
  String op = changeEvent.getOp();
  String txId = changeEvent.getSource().get("txId").asText();


  // logical decoding message
  if (op.equals("m")) {
    Message message = changeEvent.getMessage();


    // an audit metadata message -> bear in mind it
    if (message.getPrefix().equals("audit")) {
      localAuditState = new AuditState(txId, message.getContent());
      return;
    }
    else {
      out.accumulate(worth);
    }
  }
  // an information change occasion -> enrich it with the metadata
  else {
    if (txId != null && localAuditState != null) {
      if (txId.equals(localAuditState.getTxId())) {
        changeEvent.setAuditData(localAuditState.getState());
      }
    else {
      localAuditState = null;
    }
  }


  changeEvent.setTransaction(null);
  out.accumulate(mapper.writeValueAsString(changeEvent));
}

The logic is as follows:

• When the incoming occasion is of sort “m” (i.e., a logical decoding message) and it’s an audit metadata occasion, put the content material of the occasion right into a Flink worth state
• When the incoming occasion is of another sort, and we’ve saved audit state for the occasion’s transaction earlier than, enrich the occasion with that state
• When the transaction id of the incoming occasion doesn’t match what’s saved within the audit state (e.g., when a transaction was issued with out a metadata occasion initially), clear the state retailer and propagate the occasion as is

You’ll find a easy but full Flink job that runs that mapping perform in opposition to the Flink CDC connector for Postgres within the aforementioned GitHub repository. See the directions within the README for working that job, triggering some knowledge adjustments, and observing the enriched change occasions. For example, let’s contemplate the next transaction which first emits a logical decoding message with the transaction metadata (consumer identify and consumer IP deal with) after which two INSERT statements:

BEGIN;
SELECT * FROM pg_logical_emit_message(true, 'audit', '{ "consumer" : "bob@instance.com", "consumer" : "10.0.0.1" }');
INSERT INTO stock.buyer(first_name, last_name, e mail) VALUES ('Bob', 'Inexperienced', 'bob@instance.com');
INSERT INTO stock.deal with
  (customer_id, sort, line_1, line_2, zip_code, metropolis, nation)
VALUES
  (currval('stock.customer_id_seq'), 'Dwelling', '12 Most important St.', 'sdf', '90210', 'Los Angeles', 'US');
COMMIT;

The enriched change occasions, as emitted by Apache Flink, would appear like so:

{
  "op" : "c",
  "ts_ms" : 1673434483049,
  "supply" : {
    "connector" : "postgresql",
    "snapshot" : false,
    "db" : "demodb",
    "desk" : "buyer"
    "lsn" : 24023128,
    "txId" : 555,
    ...
  },
  "earlier than" : null,
  "after" : {
    "id" : 1018,
    "first_name" : "Bob",
    "last_name" : "Inexperienced",
    "e mail" : "bobasdf@instance.com"
  },
  "auditData" : {
    "consumer" : "bob@instance.com",
    "consumer" : "10.0.0.1"
  }
}
{
  "op" : "c",
  "ts_ms" : 1673434483050,
  "supply" : {
    "connector" : "postgresql",
    "snapshot" : false,
    "db" : "demodb",
    "desk" : "deal with"
    "lsn" : 24023129,
    "txId" : 555,
    ...
  },
  "earlier than" : null,
  "after" : {
    "id" : 10007,
    "customer_id" : 1018,
    "sort" : "Dwelling",
    "line_1" : "12 Most important St.",
    "line_2" : "sdf",
    "zip_code" : "90210",
    "metropolis" : "Los Angeles",
    "nation" : "US"
  },
  "auditData" : {
    "consumer" : "bob@instance.com",
    "consumer" : "10.0.0.1"
  }
}

Throughout the identical Flink job, you now might add a sink connector and as an example write the enriched occasions right into a Kafka subject. Alternatively, relying on what you are promoting necessities, it may be a good suggestion to propagate the change occasions right into a queryable retailer, as an example, an OLAP retailer like Apache Pinot or Clickhouse. You possibly can use the identical strategy for enriching change occasions with contextual metadata for different functions too, usually talking for capturing all types of “intent” which isn’t immediately continued within the enterprise tables of your software.

Bonus: Advancing Replication Slots

Lastly, let’s focus on a technical use case for logical decoding messages: advancing Postgres replication slots. This may turn out to be useful in sure eventualities, the place in any other case massive segments of the WAL may very well be retained by the database, finally inflicting the database machine to expire of disk house.

It’s because replication slots are all the time created within the context of a particular database, whereas the WAL is shared between all of the databases on the identical Postgres host. This implies a replication slot arrange for a database with none knowledge adjustments and which, due to this fact, can’t advance, will retain probably massive chunks of WAL if adjustments are made to a different database on the identical host.

To expertise this case, cease the at present working Docker Compose set-up and launch this various Compose file from the instance mission:

docker compose -f docker-compose-multi-db.yml up

This spins up a Postgres database container with two databases, DB1 and DB2. Then launch the AdvanceSlotMain class. You are able to do so by way of Maven (observe that is only for demonstration and improvement functions; often, you’d package deal up your Flink job as a JAR and deploy it to a working Flink cluster):

mvn exec:exec@advanceslot

It runs a easy Flink pipeline that retrieves all adjustments from the DB2 database and prints them out on the console. Now, do some adjustments on the DB1 database:

docker run --tty --rm -i 
  --network logical-decoding-network 
  quay.io/debezium/tooling:1.2 
  bash -c 'pgcli postgresql://postgresuser:postgrespw@order-db:5432/db1'

postgresuser@order-db:db1> CREATE TABLE knowledge (id INTEGER, worth TEXT);
postgresuser@order-db:db1> INSERT INTO knowledge SELECT generate_series(1,1000) AS id, md5(random()::textual content) AS worth;

Question the standing of the replication slot (“flink”, arrange for database “DB2”), and as you retain working extra inserts in DB1, you’ll see that the retained WAL of that slot repeatedly grows, so long as there are not any adjustments performed over in DB2:

postgresuser@order-db:db1> SELECT
  slot_name,
  database,
  pg_size_pretty(
    pg_wal_lsn_diff(
      pg_current_wal_lsn(), restart_lsn)) AS retained_wal,
  energetic,
  restart_lsn, confirmed_flush_lsn FROM pg_replication_slots;
+-----------------+------------+----------------+----------+---------------+-----------------------+
| slot_name       | database   | retained_wal   | energetic   | restart_lsn   | confirmed_flush_lsn   |
|-----------------+------------+----------------+----------+---------------+-----------------------|
| flink           | db2        | 526 kB         | True     | 0/22BA030     | 0/22BA030             |
+-----------------+------------+----------------+----------+---------------+-----------------------+

The issue is that so long as there are not any adjustments within the DB2 database, the CDC connector of the working Flink job won’t ever be invoked and thus by no means have an opportunity to acknowledge the newest processed LSN of its replication slot. Now, let’s use pg_logical_emit_message() to repair this case. Get one other Postgres shell, this time for DB2, and emit a message like so:

docker run --tty --rm -i 
  --network logical-decoding-network 
  quay.io/debezium/tooling:1.2 
  bash -c 'pgcli postgresql://postgresuser:postgrespw@order-db:5432/db2'

postgresuser@order-db:db2> SELECT pg_logical_emit_message(false, 'heartbeat', now()::varchar);

Within the console output of AdvanceSlotMain it is best to see the change occasion emitted by the Debezium connector for that message. With the following checkpoint issued by Flink (search for “Accomplished checkpoint XYZ for job …” messages within the log), the LSN of that occasion may even be flushed to the database, primarily permitting the database to discard any WAL segments earlier than that. In the event you now look at the replication slot once more, it is best to discover that the “retained WAL” worth is way decrease than earlier than (as this course of is asynchronous, it could take a bit till the disk house is freed up).

Wrapping Up

Logical decoding messages should not broadly recognized but very highly effective instruments, which needs to be within the field for each software program engineer working with Postgres. As you’ve seen, the flexibility to emit messages into the write-ahead log with out them ever surfacing in any precise desk permits for numerous attention-grabbing use instances, similar to dependable knowledge alternate between microservices (thus avoiding unsafe twin writes), software logging, or offering metadata for constructing audit logs. Using stateful stream processing utilizing Apache Flink, you possibly can enrich and route your captured messages in addition to apply different operations in your knowledge change occasions, similar to filtering, becoming a member of, windowed aggregations, and extra.

The place there may be nice energy, there are additionally nice tasks. As logical decoding messages don’t have an specific schema, in contrast to your database tables, the appliance developer should outline smart contracts and thoroughly evolve them, all the time preserving backward compatibility in thoughts. The CloudEvents format generally is a helpful basis on your customized message schemas, offering all of the producers and customers in a corporation with a constant message construction and well-defined semantics.

In the event you’d wish to get began together with your explorations round logical decoding messages, have a look at the GitHub repo accompanying this text, which accommodates the supply code of all of the examples proven above and detailed directions for working them.

Many due to Hans-Peter Grahsl, Robert Metzger, and Srini Penchikala for his or her suggestions whereas writing this text.