transaction - {@link #loadTransaction}: the Debezium {@code transaction} block of the
+ * change event ({@code id}, {@code total_order} and {@code data_collection_order}), or {@code
+ * null} unless the connector has {@code provide.transaction.metadata=true}
*
*
* Full metadata
@@ -76,6 +83,14 @@
*
* By default ({@code fullMetadata = false}) the output only contains the six keys above.
*
+ *
Transaction metadata
+ *
+ * When the connector has {@code provide.transaction.metadata=true}, Debezium also emits a {@code
+ * BEGIN} and an {@code END} event around the change events of each transaction. These have no
+ * {@code source}, {@code before} or {@code after}, so they are mapped to their own payload instead
+ * ({@link #loadTransactionEvent}): {@code status}, {@code id}, {@code ts_ms}, {@code event_count}
+ * and {@code data_collections}.
+ *
*
Usage Example
*
* Map each SourceRecord to JSON
@@ -98,10 +113,17 @@ public class SourceRecordJson {
* existing metadata key), so they are never added as full metadata extras.
*/
private static final Set RESERVED_METADATA_KEYS =
- ImmutableSet.of("connector", "version", "name", "db", "database", "schema", "table", "op");
+ ImmutableSet.of(
+ "connector", "version", "name", "db", "database", "schema", "table", "op", "transaction");
+
+ private static final String SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME =
+ "io.debezium.connector.common.TransactionMetadataKey";
+
+ private static final Gson GSON = new GsonBuilder().serializeNulls().create();
private final @Nullable Struct value;
private final @Nullable Event event;
+ private final @Nullable TransactionEvent transactionEvent;
private final Map metadataExtras;
/**
@@ -128,14 +150,21 @@ public SourceRecordJson(@Nullable SourceRecord sourceRecord, boolean fullMetadat
this.value = (Struct) sourceRecord.value();
if (this.value == null) {
- this.event = new Event(null, null, null);
+ this.event = new Event(null, null, null, null);
+ this.transactionEvent = null;
+ this.metadataExtras = Collections.emptyMap();
+ } else if (isTransactionMetadataEvent(sourceRecord.keySchema())) {
+ this.event = null;
+ this.transactionEvent = this.loadTransactionEvent();
this.metadataExtras = Collections.emptyMap();
} else {
Event.Metadata metadata = this.loadMetadata();
Event.Before before = this.loadBefore();
Event.After after = this.loadAfter();
+ Event.Transaction transaction = this.loadTransaction();
- this.event = new Event(metadata, before, after);
+ this.event = new Event(metadata, before, after, transaction);
+ this.transactionEvent = null;
this.metadataExtras =
fullMetadata && metadata != null ? this.loadMetadataExtras() : Collections.emptyMap();
}
@@ -251,12 +280,64 @@ private Event.After loadAfter() {
return new Event.After(fields);
}
+ /**
+ * Extracts the transaction field within SourceRecord, set when the connector has {@code
+ * provide.transaction.metadata=true}.
+ *
+ * @return Transaction
+ */
+ private Event.Transaction loadTransaction() {
+ @Nullable Struct transaction;
+ try {
+ transaction = (Struct) this.value.get("transaction");
+ } catch (DataException e) {
+ return null;
+ }
+ if (transaction == null) {
+ return null;
+ }
+
+ return new Event.Transaction(
+ transaction.getString("id"),
+ transaction.getInt64("total_order"),
+ transaction.getInt64("data_collection_order"));
+ }
+
+ /**
+ * Extracts the payload of a transaction BEGIN/END event.
+ *
+ * @return TransactionEvent
+ */
+ private TransactionEvent loadTransactionEvent() {
+ @Nullable List dataCollections = null;
+ @Nullable List dataCollectionStructs = this.value.getArray("data_collections");
+ if (dataCollectionStructs != null) {
+ dataCollections = new ArrayList<>();
+ for (Struct dataCollection : dataCollectionStructs) {
+ dataCollections.add(
+ new TransactionEvent.DataCollection(
+ dataCollection.getString("data_collection"),
+ dataCollection.getInt64("event_count")));
+ }
+ }
+
+ return new TransactionEvent(
+ this.value.getString("status"),
+ this.value.getString("id"),
+ this.value.getInt64("ts_ms"),
+ this.value.getInt64("event_count"),
+ dataCollections);
+ }
+
/**
* Transforms the extracted data to a JSON string.
*
* @return JSON String
*/
public String toJson() {
+ if (this.transactionEvent != null) {
+ return this.transactionEvent.toJson();
+ }
return this.event.toJson(this.metadataExtras);
}
@@ -291,6 +372,7 @@ static class Event implements Serializable {
private final SourceRecordJson.Event.Metadata metadata;
private final SourceRecordJson.Event.Before before;
private final SourceRecordJson.Event.After after;
+ private final SourceRecordJson.Event.Transaction transaction;
/**
* Event Initializer.
@@ -298,14 +380,17 @@ static class Event implements Serializable {
* @param metadata Metadata retrieved from SourceRecord
* @param before Before data retrieved from SourceRecord
* @param after After data retrieved from SourceRecord
+ * @param transaction Transaction block retrieved from SourceRecord
*/
public Event(
SourceRecordJson.Event.Metadata metadata,
SourceRecordJson.Event.Before before,
- SourceRecordJson.Event.After after) {
+ SourceRecordJson.Event.After after,
+ SourceRecordJson.Event.Transaction transaction) {
this.metadata = metadata;
this.before = before;
this.after = after;
+ this.transaction = transaction;
}
/**
@@ -325,17 +410,16 @@ public String toJson() {
* @return JSON String
*/
String toJson(Map extras) {
- Gson gson = new GsonBuilder().serializeNulls().create();
if (extras.isEmpty() || metadata == null) {
- return gson.toJson(this);
+ return GSON.toJson(this);
}
// Gson can't flatten a map into an object, so append the extras to the metadata tree.
- JsonObject tree = gson.toJsonTree(this).getAsJsonObject();
+ JsonObject tree = GSON.toJsonTree(this).getAsJsonObject();
JsonObject metadataJson = tree.getAsJsonObject("metadata");
for (Map.Entry extra : extras.entrySet()) {
- metadataJson.add(extra.getKey(), gson.toJsonTree(extra.getValue()));
+ metadataJson.add(extra.getKey(), GSON.toJsonTree(extra.getValue()));
}
- return gson.toJson(tree);
+ return GSON.toJson(tree);
}
/** Depicts the metadata within a SourceRecord. It has valuable fields. */
@@ -401,5 +485,107 @@ public After(@Nullable Map fields) {
this.fields = fields;
}
}
+
+ /** Depicts the transaction field within SourceRecord. */
+ @SuppressWarnings("unused") // Unused params are validated as JSON
+ static class Transaction implements Serializable {
+ private final @Nullable String id;
+
+ @SerializedName("total_order")
+ private final @Nullable Long totalOrder;
+
+ @SerializedName("data_collection_order")
+ private final @Nullable Long dataCollectionOrder;
+
+ /**
+ * Transaction Initializer.
+ *
+ * @param id Transaction id
+ * @param totalOrder Position of the event among all events of the transaction
+ * @param dataCollectionOrder Position of the event among the events of its data collection
+ */
+ public Transaction(
+ @Nullable String id, @Nullable Long totalOrder, @Nullable Long dataCollectionOrder) {
+ this.id = id;
+ this.totalOrder = totalOrder;
+ this.dataCollectionOrder = dataCollectionOrder;
+ }
+ }
+ }
+
+ /** Depicts a transaction BEGIN/END SourceRecord as a TransactionEvent to be mapped as JSON. */
+ @SuppressWarnings("unused") // Unused params are validated as JSON
+ static class TransactionEvent implements Serializable {
+ private final @Nullable String status;
+ private final @Nullable String id;
+
+ @SerializedName("ts_ms")
+ private final @Nullable Long tsMs;
+
+ @SerializedName("event_count")
+ private final @Nullable Long eventCount;
+
+ @SerializedName("data_collections")
+ private final @Nullable List dataCollections;
+
+ /**
+ * TransactionEvent Initializer.
+ *
+ * @param status {@code BEGIN} or {@code END}
+ * @param id Transaction id
+ * @param tsMs Time the transaction began or ended, in milliseconds
+ * @param eventCount Number of events in the transaction, only set on {@code END}
+ * @param dataCollections Number of events per data collection, only set on {@code END}
+ */
+ public TransactionEvent(
+ @Nullable String status,
+ @Nullable String id,
+ @Nullable Long tsMs,
+ @Nullable Long eventCount,
+ @Nullable List dataCollections) {
+ this.status = status;
+ this.id = id;
+ this.tsMs = tsMs;
+ this.eventCount = eventCount;
+ this.dataCollections = dataCollections;
+ }
+
+ /**
+ * Transforms the TransactionEvent to a JSON string.
+ *
+ * @return JSON String
+ */
+ public String toJson() {
+ return GSON.toJson(this);
+ }
+
+ /** Depicts an entry of the data_collections field within SourceRecord. */
+ @SuppressWarnings("unused") // Unused params are validated as JSON
+ static class DataCollection implements Serializable {
+ @SerializedName("data_collection")
+ private final @Nullable String dataCollection;
+
+ @SerializedName("event_count")
+ private final @Nullable Long eventCount;
+
+ /**
+ * DataCollection Initializer.
+ *
+ * @param dataCollection Data collection (table) name
+ * @param eventCount Number of events of the transaction in this data collection
+ */
+ public DataCollection(@Nullable String dataCollection, @Nullable Long eventCount) {
+ this.dataCollection = dataCollection;
+ this.eventCount = eventCount;
+ }
+ }
+ }
+
+ /**
+ * Whether the record is a transaction BEGIN/END event, emitted when {@code
+ * provide.transaction.metadata} is enabled. These are identified by their key schema name.
+ */
+ private static boolean isTransactionMetadataEvent(@Nullable Schema keySchema) {
+ return keySchema != null && SCHEMA_TRANSACTION_METADATA_EVENT_KEY_NAME.equals(keySchema.name());
}
}
diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java
index 29c8e33ac06d..0a3954c46287 100644
--- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java
+++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOMySqlConnectorIT.java
@@ -289,7 +289,7 @@ public void testDebeziumIOMySql() {
+ "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"addresses\"},\"before\":null,"
+ "\"after\":{\"fields\":{\"zip\":\"76036\",\"city\":\"Euless\","
+ "\"street\":\"3183 Moore Avenue\",\"id\":10,\"state\":\"Texas\",\"customer_id\":1001,"
- + "\"type\":\"SHIPPING\"}}}";
+ + "\"type\":\"SHIPPING\"}},\"transaction\":null}";
PAssert.that(results)
.satisfies(
diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java
index 0a76415d23ba..a3da4d2d497b 100644
--- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java
+++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/DebeziumIOPostgresSqlConnectorIT.java
@@ -18,10 +18,15 @@
package org.apache.beam.io.debezium;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasItem;
import io.debezium.connector.postgresql.PostgresConnector;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
import javax.sql.DataSource;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.StringUtf8Coder;
@@ -183,7 +188,7 @@ public void testDebeziumIOPostgresSql() {
"{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"beam-debezium-connector\","
+ "\"database\":\"inventory\",\"schema\":\"inventory\",\"table\":\"customers\"},\"before\":null,"
+ "\"after\":{\"fields\":{\"last_name\":\"Thomas\",\"id\":1001,\"first_name\":\"Sally\","
- + "\"email\":\"sally.thomas@acme.com\"}}}";
+ + "\"email\":\"sally.thomas@acme.com\"}},\"transaction\":null}";
PAssert.that(results)
.satisfies(
@@ -195,4 +200,83 @@ public void testDebeziumIOPostgresSql() {
p.run().waitUntilFinish();
POSTGRES_SQL_CONTAINER.stop();
}
+
+ /**
+ * Debezium - PostgresSql connector Test with transaction metadata.
+ *
+ * Tests that the transaction BEGIN/END events emitted with {@code
+ * provide.transaction.metadata=true}, which have no {@code source} block, are mapped to JSON
+ * alongside the data change events of the transaction, and that with full metadata each change
+ * event carries its {@code transaction} block.
+ */
+ @Test
+ public void testShouldProcessTransactionMetadata() throws SQLException {
+ POSTGRES_SQL_CONTAINER.start();
+
+ try (Connection connection = getPostgresDatasource().getConnection();
+ Statement statement = connection.createStatement()) {
+ // Create the slot first so it captures the commit below (no_data skips the snapshot).
+ statement.execute(
+ "SELECT pg_create_logical_replication_slot('beam_transaction_metadata', 'decoderbufs')");
+ connection.setAutoCommit(false);
+ statement.executeUpdate(
+ "INSERT INTO inventory.customers (id, first_name, last_name, email) "
+ + "VALUES (5001, 'Ann', 'One', 'ann.one@example.com')");
+ statement.executeUpdate(
+ "INSERT INTO inventory.customers (id, first_name, last_name, email) "
+ + "VALUES (5002, 'Bob', 'Two', 'bob.two@example.com')");
+ statement.executeUpdate("UPDATE inventory.customers SET last_name = 'Uno' WHERE id = 5001");
+ connection.commit();
+ }
+
+ String host = POSTGRES_SQL_CONTAINER.getHost();
+ String port = POSTGRES_SQL_CONTAINER.getMappedPort(5432).toString();
+
+ PipelineOptions options = PipelineOptionsFactory.create();
+ Pipeline p = Pipeline.create(options);
+ PCollection results =
+ p.apply(
+ DebeziumIO.readAsJson()
+ .withConnectorConfiguration(
+ DebeziumIO.ConnectorConfiguration.create()
+ .withUsername("debezium")
+ .withPassword("dbz")
+ .withConnectorClass(PostgresConnector.class)
+ .withHostName(host)
+ .withPort(port)
+ .withConnectionProperty("database.dbname", "inventory")
+ .withConnectionProperty("table.include.list", "inventory.customers")
+ .withConnectionProperty("snapshot.mode", "no_data")
+ .withConnectionProperty("provide.transaction.metadata", "true")
+ .withConnectionProperty("slot.name", "beam_transaction_metadata"))
+ .withFullMetadata(true)
+ .withMaxNumberOfRecords(5)
+ .withMaxTimeToRun(120_000L));
+
+ PAssert.that(results)
+ .satisfies(
+ (Iterable res) -> {
+ assertThat(Lists.newArrayList(res).size(), equalTo(5));
+ assertThat(res, hasItem(containsString("{\"status\":\"BEGIN\",")));
+ assertThat(res, hasItem(containsString("\"email\":\"ann.one@example.com\"")));
+ assertThat(res, hasItem(containsString("\"email\":\"bob.two@example.com\"")));
+ assertThat(
+ res,
+ hasItem(
+ allOf(
+ containsString("\"last_name\":\"Uno\""),
+ containsString("\"transaction\":{\"id\":\""),
+ containsString("\"total_order\":3,\"data_collection_order\":3}"))));
+ assertThat(
+ res,
+ hasItem(
+ containsString(
+ "\"event_count\":3,\"data_collections\":"
+ + "[{\"data_collection\":\"inventory.customers\",\"event_count\":3}]")));
+ return null;
+ });
+
+ p.run().waitUntilFinish();
+ POSTGRES_SQL_CONTAINER.stop();
+ }
}
diff --git a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java
index bcb33e97e4df..01fc6f5bd1ff 100644
--- a/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java
+++ b/sdks/java/io/debezium/src/test/java/org/apache/beam/io/debezium/SourceRecordJsonTest.java
@@ -23,6 +23,7 @@
import static org.junit.Assert.assertTrue;
import java.io.Serializable;
+import java.util.Collections;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
@@ -51,7 +52,8 @@ public void testSourceRecordJson() {
+ "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}},"
+ "\"after\":{\"fields\":{\"country\":null,\"distance\":123.423,\"birthYear\":null,"
+ "\"name\":\"after-name\","
- + "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}}}";
+ + "\"temperature\":104.4,\"childrenAndAge\":null,\"age\":16}},"
+ + "\"transaction\":null}";
assertEquals(expectedJson, jsonString);
}
@@ -76,11 +78,45 @@ public void testFullMetadata() {
.put("file", "mysql-bin.000003")
.put("pos", 796L)
.put("row", 0);
+
+ Schema transactionSchema =
+ SchemaBuilder.struct()
+ .optional()
+ .name("event.block")
+ .field("id", Schema.STRING_SCHEMA)
+ .field("total_order", Schema.INT64_SCHEMA)
+ .field("data_collection_order", Schema.INT64_SCHEMA)
+ .build();
+ Struct transaction =
+ new Struct(transactionSchema)
+ .put("id", "796:48630784")
+ .put("total_order", 1L)
+ .put("data_collection_order", 1L);
+ Schema rowSchema = SchemaBuilder.struct().optional().field("id", Schema.INT32_SCHEMA).build();
+ Struct source = postgresSource();
+ Schema envelopeSchema =
+ SchemaBuilder.struct()
+ .field("before", rowSchema)
+ .field("after", rowSchema)
+ .field("source", source.schema())
+ .field("op", Schema.STRING_SCHEMA)
+ .field("transaction", transactionSchema)
+ .build();
+ Struct envelope =
+ new Struct(envelopeSchema)
+ .put("after", new Struct(rowSchema).put("id", 1001))
+ .put("source", source)
+ .put("op", "c")
+ .put("transaction", transaction);
+ SourceRecord transactionRecord = new SourceRecord(null, null, null, envelopeSchema, envelope);
+
+ // MySQL has no "schema" source field, so the binlog "file" is also exposed as "schema".
assertEquals(
"{\"metadata\":{\"connector\":\"mysql\",\"version\":\"3.5.2.Final\",\"name\":\"n\","
+ "\"database\":\"inventory\",\"schema\":\"mysql-bin.000003\",\"table\":\"customers\","
+ "\"op\":\"c\",\"ts_ms\":1791689147000,\"gtid\":null,\"file\":\"mysql-bin.000003\","
- + "\"pos\":796,\"row\":0},\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}",
+ + "\"pos\":796,\"row\":0},\"before\":null,\"after\":{\"fields\":{\"id\":1001}},"
+ + "\"transaction\":null}",
new SourceRecordJson(buildEnvelopeRecord(mySql, "c"), true).toJson());
// PostgreSQL "schema" is a source field: it stays the existing key and isn't repeated.
@@ -88,8 +124,73 @@ public void testFullMetadata() {
"{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"n\","
+ "\"database\":\"postgres\",\"schema\":\"inventory\",\"table\":\"customers\","
+ "\"op\":\"u\",\"ts_ms\":1791689147000,\"lsn\":37094192},"
- + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}}}",
+ + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}},\"transaction\":null}",
new SourceRecordJson(buildEnvelopeRecord(postgresSource(), "u"), true).toJson());
+
+ // With provide.transaction.metadata, the envelope's transaction block is included.
+ assertEquals(
+ "{\"metadata\":{\"connector\":\"postgresql\",\"version\":\"3.5.2.Final\",\"name\":\"n\","
+ + "\"database\":\"postgres\",\"schema\":\"inventory\",\"table\":\"customers\","
+ + "\"op\":\"c\",\"ts_ms\":1791689147000,\"lsn\":37094192},"
+ + "\"before\":null,\"after\":{\"fields\":{\"id\":1001}},"
+ + "\"transaction\":{\"id\":\"796:48630784\",\"total_order\":1,"
+ + "\"data_collection_order\":1}}",
+ new SourceRecordJson(transactionRecord, true).toJson());
+ }
+
+ @Test
+ public void testTransactionMetadataEvent() {
+ Schema keySchema =
+ SchemaBuilder.struct()
+ .name("io.debezium.connector.common.TransactionMetadataKey")
+ .field("id", Schema.STRING_SCHEMA)
+ .build();
+ Schema dataCollectionSchema =
+ SchemaBuilder.struct()
+ .field("data_collection", Schema.STRING_SCHEMA)
+ .field("event_count", Schema.INT64_SCHEMA)
+ .build();
+ Schema valueSchema =
+ SchemaBuilder.struct()
+ .name("io.debezium.connector.common.TransactionMetadataValue")
+ .field("status", Schema.STRING_SCHEMA)
+ .field("id", Schema.STRING_SCHEMA)
+ .field("event_count", Schema.OPTIONAL_INT64_SCHEMA)
+ .field("data_collections", SchemaBuilder.array(dataCollectionSchema).optional().build())
+ .field("ts_ms", Schema.INT64_SCHEMA)
+ .build();
+ Struct key = new Struct(keySchema).put("id", "781:39950456");
+
+ Struct begin =
+ new Struct(valueSchema)
+ .put("status", "BEGIN")
+ .put("id", "781:39950456")
+ .put("ts_ms", 1792926401751L);
+ assertEquals(
+ "{\"status\":\"BEGIN\",\"id\":\"781:39950456\",\"ts_ms\":1792926401751,"
+ + "\"event_count\":null,\"data_collections\":null}",
+ new SourceRecordJson(new SourceRecord(null, null, null, keySchema, key, valueSchema, begin))
+ .toJson());
+
+ Struct end =
+ new Struct(valueSchema)
+ .put("status", "END")
+ .put("id", "781:39950456")
+ .put("event_count", 3L)
+ .put(
+ "data_collections",
+ Collections.singletonList(
+ new Struct(dataCollectionSchema)
+ .put("data_collection", "inventory.customers")
+ .put("event_count", 3L)))
+ .put("ts_ms", 1792926401760L);
+ assertEquals(
+ "{\"status\":\"END\",\"id\":\"781:39950456\",\"ts_ms\":1792926401760,"
+ + "\"event_count\":3,\"data_collections\":"
+ + "[{\"data_collection\":\"inventory.customers\",\"event_count\":3}]}",
+ new SourceRecordJson(
+ new SourceRecord(null, null, null, keySchema, key, valueSchema, end), true)
+ .toJson());
}
@Test
diff --git a/sdks/python/apache_beam/io/debezium.py b/sdks/python/apache_beam/io/debezium.py
index 8e7b88da687d..032a4cdde60a 100644
--- a/sdks/python/apache_beam/io/debezium.py
+++ b/sdks/python/apache_beam/io/debezium.py
@@ -149,6 +149,9 @@ def __init__(
:param connection_properties: properties of the debezium
connection passed as string
with format [propertyName=property;]*
+ Set ``provide.transaction.metadata=true``
+ to add a ``transaction`` key to records
+ and emit transaction BEGIN/END records.
:param start_offset: starting offset to resume the connector from
a previously seen position. Provided as a list
of "key=value" strings, where numeric values are