HDDS-15562. EventNotification: persistence strategy for consumer checkpoint (via Ozone filesystem bucket properties)#10515
Conversation
Persist the checkpoint within the Ozone filesystem itself so that another OM can pick up if the leader changes.
gardenia
changed the base branch from
master
to
HDDS-13513_Event_Notification_FeatureBranch
|
@ChenSammi FYI |
If the event notification checkpoint bucket does not exist (or becomes inaccessible due to permissions/disk full), SetBucketProperty OMRequests fail at the state machine level. Previously, these failures were caught but ignored, allowing the in-memory seek position to advance while no checkpoint was written. Upon OM restart, the poller loaded null and replayed all events starting from transaction 1, creating infinite duplicate event storms on Kafka. Summary of improvements: * Before polling or publishing any events, verify checkpoint storage exists via a lightweight, read-only load() call. If it fails, the poller logs a wanring, suspends any Kafka publishes, and waits to retry on the next 2-second poll. * In the OzoneFileCheckpointStrategy we propagate IOExceptions when SetBucketProperty OMRequests return a non-OK status. Inside seekPosition.set(), catching this exception sets checkpointVerified to false and blocks in-memory checkpoint advancement. This suspends any Kafka publishes on subsequent polls. * Reset saveCount to 0 upon catching a write IOException. Once the storage becomes accessible (read succeeds), the first subsequent save bypasses throttling (writes immediately) to confirm write-path accessibility before resuming normal throttled writes.
There was a problem hiding this comment.
Pull request overview
This PR introduces a persistent, HA-friendly checkpoint mechanism for OM event-notification consumers by storing the consumer checkpoint in Ozone bucket metadata (replicated via OM HA), allowing a new leader OM to resume from the last persisted position after leadership changes.
Changes:
- Add a
NotificationCheckpointStrategyabstraction and plumb it throughOMEventListenerPluginContext. - Implement
OzoneFileCheckpointStrategythat persists the checkpoint as bucket metadata (with throttled writes and failure recovery). - Update the ledger poller/publisher to load, verify accessibility of, and persist the seek position; add unit tests covering persistence and recovery behavior.
Reviewed changes
Copilot reviewed 11 out of 11 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOzoneFileCheckpointStrategy.java | Adds unit tests for checkpoint save throttling, load behavior, and failure recovery. |
| hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/OzoneManager.java | Reorders initialization to ensure plugin manager setup occurs after metadata reader is available. |
| hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OzoneFileCheckpointStrategy.java | New checkpoint strategy implementation persisting checkpoint via bucket metadata updates. |
| hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginManager.java | Constructs plugin context with a checkpoint strategy (when available). |
| hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContextImpl.java | Stores and exposes the new checkpoint strategy via the plugin context implementation. |
| hadoop-ozone/ozone-manager-plugins/src/test/java/org/apache/hadoop/ozone/om/eventlistener/TestOMEventListenerLedgerPollerRecovery.java | Adds tests validating poller suspension/resumption behavior when checkpoint access becomes unhealthy/healthy. |
| hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPollerSeekPosition.java | Updates seek position to load/save via checkpoint strategy and track verification state. |
| hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerLedgerPoller.java | Adds a pre-run checkpoint accessibility gate to avoid polling when storage is inaccessible. |
| hadoop-ozone/ozone-manager-plugins/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerKafkaPublisher.java | Wires the checkpoint strategy from plugin context into the poller seek position. |
| hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/OMEventListenerPluginContext.java | Extends the plugin context API to expose NotificationCheckpointStrategy. |
| hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/eventlistener/NotificationCheckpointStrategy.java | Introduces the checkpoint strategy interface. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| public static final int OZONE_OM_PLUGIN_CHECKPOINT_SAVE_INTERVAL_DEFAULT = 100; | ||
|
|
||
| private static final String METDATA_KEY = "notification-checkpoint"; | ||
|
|
| public OzoneFileCheckpointStrategy(OzoneManager ozoneManager, final IOmMetadataReader omMetadataReader, | ||
| OzoneConfiguration conf) { | ||
| this.ozoneManager = ozoneManager; | ||
| this.volume = conf.get(OZONE_OM_PLUGIN_CHECKPOINT_VOLUME, OZONE_OM_PLUGIN_CHECKPOINT_VOLUME_DEFAULT); |
| public void set(String val) { | ||
| LOG.debug("Setting seek position {}", val); | ||
| // NOTE: this in-memory view of the seek position needs to be kept | ||
| // up to date because the OMEventListenerLedgerPoller has a | ||
| // reference to it | ||
| seekPosition.set(val); | ||
| try { | ||
| if (checkpointStrategy != null) { | ||
| checkpointStrategy.save(val); | ||
| } |
| NotificationCheckpointStrategy checkpointStrategy = null; | ||
| if (ozoneManager != null && ozoneManager.getOmMetadataReader() != null) { | ||
| checkpointStrategy = new OzoneFileCheckpointStrategy(ozoneManager, | ||
| ozoneManager.getOmMetadataReader().get(), conf); | ||
| } |
|
@gardenia , is the idea of using a dedicated ozone file system volume and bucket to persist a progress indicator previously discussed and have the consensus? |
@ChenSammi the general concept of using the ozone filesystem as a persistence mechanism for this checkpoint position (i.e. the event position that the consumer last consumed from the event ledger and wrote out to kafka) was discussed at a prior community call and the general gist of using the Ozone filesystem as a HA persistence approach for this was suggested (IIRC) by @errose28 The specific mechanic of requiring a volume and bucket to be preconfigured (with suitable defaults but configurable names) and using the bucket properties to persist the checkpoint is one that evolved over time. The initial strawman implementation of this I had was that we would still use a "special" volume and bucket but the checkpoint position would be written to a file or stored in file metadata. However, the bucket properties approach has the virtue of requiring only a single OM request (SetBucketPropertyRequest) to distribute the write. I'm open to other suggestions. I am CC'ing @errose28 for his thoughts. |
* OzoneFileCheckpointStrategy / OMEventListenerPluginManager: - Rename METDATA_KEY typo to METADATA_KEY. - Remove the unused IOmMetadataReader constructor parameter. - Simplify OzoneFileCheckpointStrategy initialization now that the unused IOmMetadataReader parameter is removed. * OMEventListenerLedgerPollerSeekPosition: - Add a fast-path bypass in set() to avoid redundant save writes when setting the seek position to its current value. * TestOMEventListenerLedgerPollerRecovery: - Add testSetToSameValueDoesNotTriggerSave to verify that the fast-path bypass works correctly.
|
@gardenia , the progress indicator is the txID, key is "notification-checkpoint", value is txID. How about use the OM metadata table to persist this value? metadata is defined in OMDBDefinition.java, currently it's used to persist data like OM layout version, ranger ozone service version, it can perfectly save the "notification-checkpoint" -> txID data, Or we can introduce a new table to save it, just like the statefulServiceConfig table in SCM, which can be used by any feature of SCM. Second, this indicator tries to prevent OM to handle events which have already been handled before OM start or OM leader transfer, but since the event model is at least once, the receiver need to have built-in capability to handle resent event notification. So the checkpoint loading or saving fails, we should still go ahead and send events out. Checkpoint is nice to have, not mandatory, it's failure should lead to the event process task function fail. Also we need to consider if event notification feature is enabled -> checkpoint is created and saved -> event notification is disabled, who is going to clean the checkpoint. If later event notification is enabled again, what will be the impact of this uncleaned checkpoint? Third, the event framework supports multiple plugins, each plugin runs independently. The current checkpoint and #10516 are specific to Kafka. If there is another plugin loaded, this plugin may also has it's checkpoint, and it needs to coordinate with Kafka about the purge, in case one plugin moving fast, while another one moving relatively slow. It looks like the current framework doesn't support this if we really want to support multiple plugins running together. BTW, I turned https://issues.apache.org/jira/browse/HDDS-13513 into an umbrella JIRA, and converted all existing event notification JIRA as the sub JIRA of HDDS-13513. Let's create new sub JIRAs of HDDS-13513 in future, so that we can easily track all event notification patches under one page. |
@ChenSammi thanks for your feedback/thoughtful suggestions. I liked the idea that using something like SetBucketPropertyRequest meant we can piggyback on existing HA storage and ratis requests but I can see how your approach would be cleaner. If we went with your suggestion I'm assuming we would need a need a new Ratis request/response to distribute this to other OMs (similar to OMSetRangerServiceVersionRequest) ? e.g. SaveCheckpointRequest ? Or is there some existing ratis request we could piggy back on for that? |
gardenia
changed the title
@ChenSammi I have opened a new PR which intends to implement the strategy you suggest for using RocksDB to store the checkpoint and address the other issues you mentioned (different checkpoint names per plugin and progress not stalled by a checkpoint write failure): |
|
superseded by #10515 |
Please describe your PR in detail:
What is the link to the Apache JIRA
https://issues.apache.org/jira/browse/HDDS-15562
How was this patch tested?
Unit tests