You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "eskabetxe (via GitHub)" <gi...@apache.org> on 2026/04/29 22:13:55 UTC
[PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
eskabetxe opened a new pull request, #4387:
URL: https://github.com/apache/flink-cdc/pull/4387
### What is the purpose of the change
Enable Postgres logical decoding messages produced by pg_logical_emit_message(transactional, prefix, content) to flow through to the user's DebeziumDeserializationSchema. Today these records are silently dropped before reaching deserialization. See [FLINK-XXXXX] for the problem statement and root cause.
### Brief change log
Base extension hooks (no domain logic; pure extensibility):
- DataSourceDialect: new default createStreamFetcher(FetchTask.Context, int) factory returning the standard IncrementalSourceStreamFetcher. Existing dialects inherit unchanged behavior.
- IncrementalSourceSplitReader.getStreamFetcher(): now obtains the fetcher via dataSourceDialect.createStreamFetcher(...) instead of direct instantiation.
- IncrementalSourceStreamFetcher.shouldEmit(): visibility private → protected so dialect subclasses can override.
- IncrementalSourceRecordEmitter.updateStreamSplitState(): visibility private → protected for the same reason.
Postgres connector:
- New PostgresSourceStreamFetcher extends the base; overrides shouldEmit() to early-return true for op="m" records when the feature flag is enabled, otherwise delegates to super. Also exposes a package-private isLogicalMessage(SourceRecord) helper.
- PostgresDialect.createStreamFetcher() returns PostgresSourceStreamFetcher with the flag read from PostgresSourceConfig.
- PostgresSourceRecordEmitter.processElement() adds a branch for logical messages: updates the stream split state and emits to deserialization. Falls through to super otherwise.
Configuration:
- New option scan.logical-message.enabled (boolean, default false) on PostgresSourceOptions. Default preserves existing behavior.
- Plumbed through PostgresSourceConfig (field + getter), PostgresSourceConfigFactory (setter), and PostgresSourceBuilder.includeLogicalMessages(boolean) (DataStream API).
### Design notes
- Why a dialect factory hook instead of an isLogicalMessage() guard in the base classes? op="m" is only produced by the Postgres connector in this repo. Putting the check in shared code would leak Postgres-specific semantics into flink-cdc-base. The factory hook is a generic extensibility point; the actual logic stays in the Postgres module.
- Why not override isDataChangeRecord() in the Postgres FetchTask.Context? It's semantically wrong — a logical message is a data change. It would also affect SourceReaderMetrics and other call sites, hiding logical messages from telemetry.
- Content is emitted as-is. pg_logical_emit_message accepts bytea — arbitrary binary content. The content field is delivered as raw byte[] (Kafka Connect Schema.BYTES). The base64-looking output you'll see in JSON-converted records is the standard JsonConverter encoding for BYTES; Avro/Protobuf converters carry raw bytes. Interpretation belongs in the user's deserializer (they know what their prefix means), not the connector.
- Default is false. Existing pipelines see no behavior change. Opt-in feature.
### Verifying this change
Added tests under flink-connector-postgres-cdc/src/test/java/.../source/reader/:
- PostgresSourceStreamFetcherTest — unit tests for isLogicalMessage() covering: op="m" (true); op="c"/"u"/"d"/"r"/"t" (false); null value; Struct without op field; Struct with null op value; non-Struct value.
- PostgresSourceStreamFetcherITCase — testcontainer-based integration test (PG 14, pgoutput, Debezium 1.9.8):
1. logicalMessagesAreEmittedWhenEnabled: with flag on, calls pg_logical_emit_message(false, 'cdc-test', 'hello') and polls until an op="m" record appears.
2. logicalMessagesAreDroppedWhenDisabled: with flag off, calls the same followed by a marker INSERT. Polls until the marker arrives (proving the WAL has advanced past the message), then asserts no op="m" record was seen.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3303728997
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherITCase.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.assigner.StreamSplitAssigner;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+import org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
+import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask;
+import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory;
+import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
+
+import io.debezium.data.Envelope;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+
+import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.isHeartbeatEvent;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration test for {@link PostgresSourceStreamFetcher} verifying that {@code
+ * pg_logical_emit_message} records reach the polled records when the feature flag is enabled and
+ * are dropped when it is disabled.
+ */
+class PostgresSourceStreamFetcherITCase extends PostgresTestBase {
Review Comment:
Lake of test e2e( which is included record emitter), such as `org.apache.flink.cdc.connectors.postgres.source.reader.PostgresSourceReaderTest`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4553051741
> @leonardBang @eskabetxe I have modified some code and test, please help review.
@loserwang1024 for @leonardBang I think we are missing some changes. I added it [here](https://github.com/eskabetxe/flink-cdc/pull/1/changes)
so it won't affect the changes you already made, if you think that they are ok I can merge the code in this PR
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3303585024
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
Review Comment:
I don't think this is a good choice to a new subclass PostgresSourceStreamFetcher here. Maybe we can move `shouldEmit` to `FetchTask.Context(such as isDataChangeRecord)`.
When we design context and dialect control the different behaviors for different databases. The IncrementalSourceStreamFetcher shares same logical.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3308108620
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
Review Comment:
@eskabetxe @loserwang1024 I'd suggest following the `FetchTask.Context` approach rather than introducing a PostgresSourceStreamFetcher subclass. Two reasons:
1. Better cohesion. The Context abstraction already serves as the dialect-specific extension point — `isDataChangeRecord()` lives there for exactly this reason. Adding `shouldEmit()` to Context keeps all Postgres-specific decision logic in one place (`PostgresSourceFetchTaskContext`), rather than splitting it between a new StreamFetcher subclass and the existing context.
2. Avoids a double-check path. The current implementation introduces `shouldEmit(record) && context.shouldEmit(record)`, which creates ambiguity about which layer owns the emit decision. If we consolidate everything into `Context.shouldEmit()`, the base `IncrementalSourceStreamFetcher` would simply call `context.shouldEmit(record)` with no subclass override needed:
```java
// IncrementalSourceStreamFetcher — no subclass needed
if (context.shouldEmit(record)) {
// emit
}
// PostgresSourceFetchTaskContext — single decision point
@Override
public boolean shouldEmit(SourceRecord record) {
if (isLogicalMessage(record)) {
return matchesPrefix(record, prefixes);
}
return super.shouldEmit(record);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3307006188
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
Review Comment:
I was doing that as logical messages are something specific to postgres.. but lets see if the change I make is more aligned with what you are asking
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4535630565
@loserwang1024 @leonardBang could you check
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4388280229
> * scan.logical-message.enabled (boolean, default false)
Hi @loserwang1024
its opt-in with scan.logical-message.enabled (boolean, default false)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4384787966
@eskabetxe , `pg_logical_emit_message` is more tightly coupled with the user's business logic rather than being a CDC log generated by PostgreSQL based on DML operations. Personally, I am skeptical about whether it should be introduced at all. @leonardBang , WDYT?
If it is introduced, what happens if the messages emitted via pg_logical_emit_message do not align with the table schema?If you insist on introducing it, please add a configuration option to explicitly enable it. Do not alter the existing logic to avoid disrupting currently running jobs that are working as expected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4396811765
> @eskabetxe , `pg_logical_emit_message` is more tightly coupled with the user's business logic rather than being a CDC log generated by PostgreSQL based on DML operations. Personally, I am skeptical about whether it should be introduced at all. @leonardBang , WDYT?
>
> If it is introduced, what happens if the messages emitted via pg_logical_emit_message do not align with the table schema?If you insist on introducing it, please add a configuration option to explicitly enable it. Do not alter the existing logic to avoid disrupting currently running jobs that are working as expected.
+1 to introduce a config and only expose it in DataStream API. We should always disabled this feature for SQL&YAML API as it will crush the entire pipeline once we delivered this kind of event, WDYT？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3307995481
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+ private final boolean includeLogicalMessages;
+
+ public PostgresSourceStreamFetcher(
+ FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+ super(taskContext, subtaskId);
+ this.includeLogicalMessages = includeLogicalMessages;
+ }
+
+ @Override
+ protected boolean shouldEmit(SourceRecord sourceRecord) {
+ if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+ return true;
+ }
+ return super.shouldEmit(sourceRecord);
+ }
+
+ static boolean isLogicalMessage(SourceRecord record) {
+ if (record.value() instanceof Struct) {
+ Struct struct = (Struct) record.value();
Review Comment:
@leonardBang @ruanhang1993 help to see the config `logical-message.prefixes`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3308108620
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
Review Comment:
@eskabetxe @loserwang1024 I'd suggest following the `FetchTask.Context` approach rather than introducing a PostgresSourceStreamFetcher subclass. Two reasons:
1. Better cohesion. The Context abstraction already serves as the dialect-specific extension point — `isDataChangeRecord()` lives there for exactly this reason. Adding `shouldEmit()` to Context keeps all Postgres-specific decision logic in one place (`PostgresSourceFetchTaskContext`), rather than splitting it between a new StreamFetcher subclass and the existing context.
2. Avoids a double-check path. The current implementation introduces `shouldEmit(record) && context.shouldEmit(record)`, which creates ambiguity about which layer owns the emit decision. If we consolidate everything into `Context.shouldEmit()`, the base `IncrementalSourceStreamFetcher` would simply call `context.shouldEmit(record)` with no subclass override needed:
```java
> // IncrementalSourceStreamFetcher — no subclass needed
> if (context.shouldEmit(record)) {
> // emit
> }
>
> // PostgresSourceFetchTaskContext — single decision point
> @Override
> public boolean shouldEmit(SourceRecord record) {
> if (isLogicalMessage(record)) {
> return matchesPrefix(record, prefixes);
> }
> return super.shouldEmit(record);
> }
>
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3303625817
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+ private final boolean includeLogicalMessages;
+
+ public PostgresSourceStreamFetcher(
+ FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+ super(taskContext, subtaskId);
+ this.includeLogicalMessages = includeLogicalMessages;
+ }
+
+ @Override
+ protected boolean shouldEmit(SourceRecord sourceRecord) {
+ if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+ return true;
+ }
+ return super.shouldEmit(sourceRecord);
+ }
+
+ static boolean isLogicalMessage(SourceRecord record) {
+ if (record.value() instanceof Struct) {
+ Struct struct = (Struct) record.value();
Review Comment:
A minor confusion regarding PostgreSQL: The WAL logs in PostgreSQL are substituted based on table names. However, the logical messages being read now seem to lack identity identifiers, making it impossible to distinguish between subscriptions (i.e., different jobs would all read the same data).
I have searched how to use logical message, it seems that prefix is used for it.
```sql
pg_logical_emit_message ( transactional boolean , prefix text , content text [, flush boolean DEFAULTfalse ] )
```
Therefore, I do think we need to add a param to control which prefix is needed to filter the target logical message.
the design of this config is the key of this issue, @leonardBang , WDYT?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3306823825
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceRecordEmitter.java:
##########
@@ -43,6 +45,18 @@ public PostgresSourceRecordEmitter(
offsetFactory);
}
+ @Override
+ protected void processElement(
+ SourceRecord element, SourceOutput<T> output, SourceSplitState splitState)
+ throws Exception {
+ if (PostgresSourceStreamFetcher.isLogicalMessage(element)) {
Review Comment:
moved to PostgresSourceRecordUtils as it is more related to sources.. but if you think will be better on another I can change
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3303585024
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
Review Comment:
I don't think this is a good choice to a new subclass PostgresSourceStreamFetcher here. Maybe we can move shouldEmit to FetchTask.Context(such as isDataChangeRecord).
When we design context and dialect control the different behaviors for different databases. The IncrementalSourceStreamFetcher shares same logical.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceRecordEmitter.java:
##########
@@ -43,6 +45,18 @@ public PostgresSourceRecordEmitter(
offsetFactory);
}
+ @Override
+ protected void processElement(
+ SourceRecord element, SourceOutput<T> output, SourceSplitState splitState)
+ throws Exception {
+ if (PostgresSourceStreamFetcher.isLogicalMessage(element)) {
Review Comment:
PostgresSourceStreamFetcher is not a util class. Please move to `PostgresTypeUtils` or `PostgresQueryUtils` or a new `PostgresSourceRecordUtils`.
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+ private final boolean includeLogicalMessages;
+
+ public PostgresSourceStreamFetcher(
+ FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+ super(taskContext, subtaskId);
+ this.includeLogicalMessages = includeLogicalMessages;
+ }
+
+ @Override
+ protected boolean shouldEmit(SourceRecord sourceRecord) {
+ if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+ return true;
+ }
+ return super.shouldEmit(sourceRecord);
+ }
+
+ static boolean isLogicalMessage(SourceRecord record) {
+ if (record.value() instanceof Struct) {
+ Struct struct = (Struct) record.value();
Review Comment:
A minor confusion regarding PostgreSQL: The WAL logs in PostgreSQL are substituted based on table names. However, the logical messages being read now seem to lack identity identifiers, making it impossible to distinguish between subscriptions (i.e., different jobs would all read the same data).
I have searched how to use logical message, it seems that prefix is used for it.
```sql
pg_logical_emit_message ( transactional boolean , prefix text , content text [, flush boolean DEFAULTfalse ] )
```
Therefore, I do think we need to add a param to control which prefix is needed to filter the target logical message. It's the key of this issue, @leonardBang , WDYT?
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java:
##########
@@ -98,6 +98,16 @@ public class PostgresSourceOptions extends JdbcSourceOptions {
+ "(1) PUBLICATION must be created beforehand with parameter publish_via_partition_root=true\n"
+ "(2) Table list (regex or predefined list) should only match the parent table name, if table list matches both parent and child tables, snapshot data will be read twice.");
+ public static final ConfigOption<Boolean> SCAN_LOGICAL_MESSAGE_ENABLED =
+ ConfigOptions.key("scan.logical-message.enabled")
Review Comment:
we need also to tell the use in this config description and document:
1. which version of postgres supports messages in pgoutput.
2. how to enable it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4359779909
@GOODBOY008 can you help review this?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4552789703
@leonardBang @eskabetxe I have modified some code and test, please help review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3308121695
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+ private final boolean includeLogicalMessages;
+
+ public PostgresSourceStreamFetcher(
+ FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+ super(taskContext, subtaskId);
+ this.includeLogicalMessages = includeLogicalMessages;
+ }
+
+ @Override
+ protected boolean shouldEmit(SourceRecord sourceRecord) {
+ if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+ return true;
+ }
+ return super.shouldEmit(sourceRecord);
+ }
+
+ static boolean isLogicalMessage(SourceRecord record) {
+ if (record.value() instanceof Struct) {
+ Struct struct = (Struct) record.value();
Review Comment:
Regarding the configuration design: if providing logical message prefixes is mandatory for PG logical decoding to work correctly (i.e., every job that writes pg_logical_emit_message must specify a prefix to differentiate subscriptions), I'd suggest we can remove *scan.logical-message.enabled* entirely and use the presence of *scan.logical-message.prefixes* as the implicit enablement signal.
When *scan.logical-message.prefixes* is set (non-empty), the feature is implicitly enabled — the user clearly intends to consume logical messages
When *scan.logical-message.prefixes* is empty or absent, no logical messages are consumed — default safe behavior preserved
This eliminates the risk of misconfiguration where enabled=true but prefixes is accidentally left empty (which would consume ALL logical messages across subscriptions — a potentially dangerous default)
It removes one config knob without losing expressiveness.
```java
// PostgresSourceFetchTaskContext.shouldEmit()
if (isLogicalMessage(record)) {
// Feature is implicitly enabled when prefixes are configured
return logicalMessagePrefixes.isEmpty() ? false
: logicalMessagePrefixes.stream().anyMatch(prefix -> matchesPrefix(record, prefix));
}
```
The PostgresSourceBuilder API could expose a single method:
```java
builder.includeLogicalMessages(List<String> prefixes)
```
where passing a non-empty list enables the feature, and omitting the call or passing an empty list keeps it disabled.
WDYT? @loserwang1024 @eskabetxe
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3306831894
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/config/PostgresSourceOptions.java:
##########
@@ -98,6 +98,16 @@ public class PostgresSourceOptions extends JdbcSourceOptions {
+ "(1) PUBLICATION must be created beforehand with parameter publish_via_partition_root=true\n"
+ "(2) Table list (regex or predefined list) should only match the parent table name, if table list matches both parent and child tables, snapshot data will be read twice.");
+ public static final ConfigOption<Boolean> SCAN_LOGICAL_MESSAGE_ENABLED =
+ ConfigOptions.key("scan.logical-message.enabled")
Review Comment:
changed
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherITCase.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.assigner.StreamSplitAssigner;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+import org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
+import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask;
+import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory;
+import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
+
+import io.debezium.data.Envelope;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+
+import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.isHeartbeatEvent;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration test for {@link PostgresSourceStreamFetcher} verifying that {@code
+ * pg_logical_emit_message} records reach the polled records when the feature flag is enabled and
+ * are dropped when it is disabled.
+ */
+class PostgresSourceStreamFetcherITCase extends PostgresTestBase {
Review Comment:
added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3306830397
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/main/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcher.java:
##########
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.cdc.connectors.base.source.reader.external.FetchTask;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+
+import io.debezium.data.Envelope;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+
+/**
+ * Stream fetcher for Postgres. When {@code includeLogicalMessages} is enabled, lets {@code
+ * pg_logical_emit_message} records (op="m") bypass table-based watermark filtering, since logical
+ * messages are not bound to a table.
+ */
+public class PostgresSourceStreamFetcher extends IncrementalSourceStreamFetcher {
+
+ private final boolean includeLogicalMessages;
+
+ public PostgresSourceStreamFetcher(
+ FetchTask.Context taskContext, int subtaskId, boolean includeLogicalMessages) {
+ super(taskContext, subtaskId);
+ this.includeLogicalMessages = includeLogicalMessages;
+ }
+
+ @Override
+ protected boolean shouldEmit(SourceRecord sourceRecord) {
+ if (includeLogicalMessages && isLogicalMessage(sourceRecord)) {
+ return true;
+ }
+ return super.shouldEmit(sourceRecord);
+ }
+
+ static boolean isLogicalMessage(SourceRecord record) {
+ if (record.value() instanceof Struct) {
+ Struct struct = (Struct) record.value();
Review Comment:
added a prefix filtering in same way we have tablesInclued
logical-message.prefixes
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on code in PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#discussion_r3306832248
##########
flink-cdc-connect/flink-cdc-source-connectors/flink-connector-postgres-cdc/src/test/java/org/apache/flink/cdc/connectors/postgres/source/reader/PostgresSourceStreamFetcherITCase.java:
##########
@@ -0,0 +1,224 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.connectors.postgres.source.reader;
+
+import org.apache.flink.api.connector.source.mocks.MockSplitEnumeratorContext;
+import org.apache.flink.cdc.connectors.base.options.StartupOptions;
+import org.apache.flink.cdc.connectors.base.source.assigner.StreamSplitAssigner;
+import org.apache.flink.cdc.connectors.base.source.meta.split.SourceRecords;
+import org.apache.flink.cdc.connectors.base.source.meta.split.StreamSplit;
+import org.apache.flink.cdc.connectors.base.source.reader.external.IncrementalSourceStreamFetcher;
+import org.apache.flink.cdc.connectors.postgres.PostgresTestBase;
+import org.apache.flink.cdc.connectors.postgres.source.PostgresDialect;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfig;
+import org.apache.flink.cdc.connectors.postgres.source.config.PostgresSourceConfigFactory;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresSourceFetchTaskContext;
+import org.apache.flink.cdc.connectors.postgres.source.fetch.PostgresStreamFetchTask;
+import org.apache.flink.cdc.connectors.postgres.source.offset.PostgresOffsetFactory;
+import org.apache.flink.cdc.connectors.postgres.testutils.UniqueDatabase;
+
+import io.debezium.data.Envelope;
+import io.debezium.relational.TableId;
+import io.debezium.relational.history.TableChanges;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.source.SourceRecord;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Predicate;
+
+import static org.apache.flink.cdc.connectors.base.utils.SourceRecordUtils.isHeartbeatEvent;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration test for {@link PostgresSourceStreamFetcher} verifying that {@code
+ * pg_logical_emit_message} records reach the polled records when the feature flag is enabled and
+ * are dropped when it is disabled.
+ */
+class PostgresSourceStreamFetcherITCase extends PostgresTestBase {
Review Comment:
added PostgresLogicalMessageE2ETest
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "loserwang1024 (via GitHub)" <gi...@apache.org>.
loserwang1024 merged PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-39582][postgres] Allow logical messages [flink-cdc]
Posted by "eskabetxe (via GitHub)" <gi...@apache.org>.
eskabetxe commented on PR #4387:
URL: https://github.com/apache/flink-cdc/pull/4387#issuecomment-4445118305
@leonardBang this is already configured with a config.. must activated it with "scan.logical-message.enabled" (default false)
I agree that this is more for the use case of using the connector directly, but is a valid configuration for that.
For the YAML app it will be disable and should not have any option to activate..
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org