You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "shibd (via GitHub)" <gi...@apache.org> on 2025/07/03 13:16:09 UTC
[PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
shibd opened a new pull request, #24484:
URL: https://github.com/apache/pulsar/pull/24484
### Motivation
The Debezium DatabaseHistory mechanism requires a complete and immutable log of all DDL changes to correctly reconstruct table schemas upon connector restart. The current PulsarDatabaseHistory implementation stores this history in a Pulsar topic but does not enforce a retention policy.
### Modifications
Create a durable subscription and never receive/ack message to make sure history topic data is retained. The data is actually very small. Additionally, we can also flexibly control this subscription's cursor to fix or skip some history data.
### Verifying this change
- You can reproduce this issue by testing: `PulsarDatabaseHistoryTest`
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: <!-- ENTER URL HERE -->
<!--
After opening this PR, the build in apache/pulsar will fail and instructions will
be provided for opening a PR in the PR author's forked repository.
apache/pulsar pull requests should be first tested in your own fork since the
apache/pulsar CI based on GitHub Actions has constrained resources and quota.
GitHub Actions provides separate quota for pull requests that are executed in
a forked repository.
The tests will be run in the forked repository until all PR review comments have
been handled, the tests pass and the PR is approved by a reviewer.
-->
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd closed pull request #24484: [fix][io] Ensure infinite retention for Debezium history topic
URL: https://github.com/apache/pulsar/pull/24484
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2183878072
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -100,6 +100,8 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
DatabaseHistory.NAME,
READER_CONFIG);
+ protected static final String DURABLE_SUB_NAME = "durable-sub";
Review Comment:
It might also confuse user if they set a specific data retention.
Is it better to use retention policy.
For example, if the topic is not created, the connector can create the topic with infinite data retention by default.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2183994526
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -100,6 +100,8 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
DatabaseHistory.NAME,
READER_CONFIG);
+ protected static final String DURABLE_SUB_NAME = "durable-sub";
Review Comment:
Under the current functionality, this topic must have infinite retention. Allowing users to set a custom data retention policy could lead to incomplete data and recovery failures.
Furthermore, if we were to manage this using data retention policies, it would introduce a dependency on pulsar-admin and likely change the permission requirements for the user when creating a connector.
Therefore, using a durable subscription is a simple and effective solution.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2184225104
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -100,6 +100,8 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
DatabaseHistory.NAME,
READER_CONFIG);
+ protected static final String DURABLE_SUB_NAME = "durable-sub";
Review Comment:
hi, @nodece thanks for the reminder, you're right.
Unfortunately, for the connector to use pulsar-admin, it would require more interface support. Furthermore, we typically only need to grant `produce` and `consume` permissions to create a connector, but this approach would also require users to grant `admin` permissions.
I think a more direct and simple solution is to modify the documentation and tell users to manually set the retention policy before they create the Debezium source connector.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2182762830
##########
pulsar-io/debezium/core/src/test/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistoryTest.java:
##########
@@ -258,8 +262,9 @@ public void testSubscriptionName() throws Exception {
assertTrue(history.exists());
try (Reader<String> ignored = history.createHistoryReader()) {
List<String> subscriptions = admin.topics().getSubscriptions(topicName);
- assertEquals(subscriptions.size(), 1);
+ assertEquals(subscriptions.size(), 2);
Review Comment:
JUnit’s `assertEquals` signature is `assertEquals(expected, actual)`. Swap the arguments to `assertEquals(2, subscriptions.size())` for clearer failure messages.
```suggestion
assertEquals(2, subscriptions.size());
```
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -165,6 +167,15 @@ public void initializeStorage() {
log.error("Failed to initialize storage", pce);
throw new RuntimeException("Failed to initialize storage", pce);
}
+ try (org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(DURABLE_SUB_NAME).topic(topicName).subscribe()) {
Review Comment:
By default, a new subscription starts at the LATEST position, so older messages won’t count toward backlog retention. Explicitly add `.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)` to ensure the durable cursor begins at the start of the topic.
```suggestion
.subscriptionName(DURABLE_SUB_NAME)
.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
.topic(topicName).subscribe()) {
```
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2182764504
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -165,6 +167,15 @@ public void initializeStorage() {
log.error("Failed to initialize storage", pce);
throw new RuntimeException("Failed to initialize storage", pce);
}
+ try (org.apache.pulsar.client.api.Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+ .subscriptionName(DURABLE_SUB_NAME).topic(topicName).subscribe()) {
Review Comment:
No need to set initial position
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Ensure infinite retention for Debezium history topic [pulsar]
Posted by "nodece (via GitHub)" <gi...@apache.org>.
nodece commented on code in PR #24484:
URL: https://github.com/apache/pulsar/pull/24484#discussion_r2184199943
##########
pulsar-io/debezium/core/src/main/java/org/apache/pulsar/io/debezium/PulsarDatabaseHistory.java:
##########
@@ -100,6 +100,8 @@ public final class PulsarDatabaseHistory extends AbstractDatabaseHistory {
DatabaseHistory.NAME,
READER_CONFIG);
+ protected static final String DURABLE_SUB_NAME = "durable-sub";
Review Comment:
@shibd If message TTL and retention are already configured on the namespace level, your change is meaningless. To achieve infinite retention, it's recommended to use pulsar-admin and set the retention value to 0, which disables retention.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org