   [trigger state state-event]
   (:fire? state))
 
+(defn watermark-init-locals [{:keys [trigger/delay]}]
+  {:delay (if delay (apply to-standard-units delay) 0)})
+
 (defn watermark-fire?
-  [{:keys [trigger/period]} trigger-state {:keys [event-type upper-bound watermarks] :as state-event}]
+  [_ _ {:keys [event-type upper-bound watermarks trigger-state] :as state-event}]
   (or (= :job-completed event-type) 
       (and (= :watermark event-type)
-           (let [standard (if period (apply to-standard-units period) 0)]
-             (> (:input watermarks) (+ upper-bound standard))))))
+           (> (:input watermarks) (+ upper-bound (:delay trigger-state))))))
 
 (defn percentile-watermark-fire?
-  [trigger trigger-state {:keys [lower-bound upper-bound event-type segment window]}]
+  [trigger _ {:keys [lower-bound upper-bound event-type segment window]}]
   ;; If this was stimulated by a new segment, check if it should fire.
   ;; Otherwise if this was a completed task, always fire.
   (or (and segment (exceeds-percentile-watermark? window trigger lower-bound upper-bound segment))

  [trigger]
  {:fire? false})

(defn percentile-watermark-init-state
  [trigger]
  ;; Intentionally return nil - this trigger is stateless.
  )

;; Init local functions

(defn segment-init-locals [trigger]

   :trigger/trigger-fire? punctuation-fire?})

(def ^:export watermark
  {:trigger/init-locals watermark-init-locals
   :trigger/trigger-fire? watermark-fire?})

(def ^:export percentile-watermark
  {:trigger/init-locals percentile-watermark-init-locals
   :trigger/trigger-fire? percentile-watermark-fire?})
