---
schema_version: "1"
id: b6543eb02b790d6b73de7956740bd389fa87ad35d52aa5fe45719277683591ad
title: Set Up Watermarks with an Object Store Example
collection: mulesoft
version: current
locale: en-us
portal_url: https://docs.salesforce.com/mulesoft/current/en-us/object-store-connector/1.2/object-store-to-watermark.md
canonical_url: https://docs.mulesoft.com/object-store-connector/1.2/object-store-to-watermark
source_site: docs.mulesoft.com
content_type: documentation
content_hash: bef98d0f6b4125f046ec13cb85ae9b04c61812c7337dd41f045482d680937984
description: MuleSoft Documentation Site
release: "1.2"
latest: false
---
---
product: Object Store Connector
version: 1.2
keywords: ObjectStore, watermark, synchronization
---

# Set Up Watermarks with an Object Store Example

> For the full documentation index, see: https://docs.mulesoft.com/llms.txt

Watermarking is a technique for storing and retrieving the point at which a periodic synchronization
should resume the next time it’s executed. Watermarking is a common use of the object store.

## Background

For an explanation of what watermarks are, see
[this blog post](https://blogs.mulesoft.com/dev-guides/how-to-tutorials/data-synchronizing-made-easy-with-mule-watermarks/).

Limitations of the earlier approach:

- Although the `<watermark/>` component allowed setting a custom `ObjectStore`, there was no easy
  way to create it. This issue is resolved in current implementations.

- The `<watermark/>` element required the `<poll/>` component (replaced by `<scheduler/>`) to return
  an iterable payload, and you had to fully iterate over that payload to obtain the new watermark
  value. Failure to do so resulted in the value being wrongfully updated or not updated at all.

- There was no way to update the watermark with a contingency value if the flow threw an error. How
  and when the value got updated was not completely clear.

When we sat down to figure out how to improve this use case, we realized that the real root of the
problem was that in reality the <watermark> shouldn’t be necessary at all. It’s existence was just a
consequence of limitations in other components:

- The watermark’s artificial iteration was just a way to compensate for the fact that many data sets
  could only be iterated once, especially when dealing with binary streams or with connectors doing
  auto paging. This problem is now fixed in Mule 4 thanks to the repeatable streaming feature

- The component’s automatic handling of the `ObjectStore` was just to compensate for UX problems on
  the old `ObjectStore` support. If the new connector were to provide easier semantics, it wouldn’t
  be needed at all.

## Mule 4 Example

This example shows how to perform the use case described in the blog.

```xml
<os:object-store name="watermarkStore" persistent="true"/>

<flow name="watermark">
    <os:retrieve key="watermark" objectStore="watermarkStore" target="watermark">
        <os:default-value>2017-09-11T00:00:00.000Z</os:default-value>
    </os:retrieve>
    <sfdc:query config-ref="config">
        <sfdc:salesforce-query>
            <![CDATA[
                #["Select Id, Name, BillingCity,Phone,Website,LastModifiedDate from account WHERE LastModifiedDate > " ++ vars.watermark]
           ]]>
       </sfdc:salesforce-query>
    </sfdc:query>
    <flow-ref name="doYourIntegrationLogic" />
    <os:store key="watermark" failIfPresent="false"
     failOnNullValue="false" objectStore="watermarkStore">
        <os:value>#[max(payload map $.LastModifiedDate)]</os:value>
    </os:store>
</flow>
```

The example is a generic flow that does not require a specific type of trigger. Though you might
invoke it from another flow, or you might use a `<scheduler/>` to trigger the flow, you can produce
watermark semantics without forcing any type of trigger.

Then the example defines a custom `ObjectStore` for the watermark, but it is also possible to use
the `ObjectStore` that is available implicitly to every app. So you do not need to define a custom
`ObjectStore`.

The flow starts by retrieving the last watermark value. The Retrieve operation (`<os:retrieve/>`)
specifies a default value (`<os:default-value/>`) that is available the first time the flow runs,
before the first watermark is calculated. This configuration avoids the need for a
`<choice> → <contains>` type of pattern.

Note that the Retrieve operation stores the watermark value in a target value (`target="watermark"`)
to prevent the connector from overriding the message payload. This makes the flow reusable and
easier to maintain. You can safely add any kind of operation before the Retrieve operation.

The query (`<sfdc:query/>`) uses the watermark (`vars.watermark`) to filter the obtained results.

The `<flow-ref />` points to another flow that does whatever processing is needed, for example,
sending data to another system, performing a transformation, or something else.

The last step (in `<os:store/>`) is to update the watermark value:

- The new watermark value should be the maximum updated timestamp. The example uses `<os:value/>` to
  provide a DataWeave expression that obtains that value.

  Note that the repeatable streaming feature prevents the need for you to know the integration
  logic. Even though the query operation auto-paged, this feature guarantees that you can still
  iterate the result set again.

- `failIfPresent` is set to `false` so that the value will be updated if it is present. Otherwise,
  the operation would only work the first time and fail if the watermark were already in the object
  store.

- `failOnNullValue` is set to `false`, so if the query comes back empty, the DataWeave expression
  (which looks for the max value) will return null. Setting it to `false` makes the connector skip
  null values, which prevents the need to use a <choice/> router to check the condition, while using
  the default value of `true` makes the connector throw a `NULL_VALUE` error.