001/* 002 * (C) Copyright 2017 Nuxeo SA (http://nuxeo.com/) and others. 003 * 004 * Licensed under the Apache License, Version 2.0 (the "License"); 005 * you may not use this file except in compliance with the License. 006 * You may obtain a copy of the License at 007 * 008 * http://www.apache.org/licenses/LICENSE-2.0 009 * 010 * Unless required by applicable law or agreed to in writing, software 011 * distributed under the License is distributed on an "AS IS" BASIS, 012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 013 * See the License for the specific language governing permissions and 014 * limitations under the License. 015 * 016 * Contributors: 017 * bdelbosc 018 */ 019package org.nuxeo.lib.stream.computation.internals; 020 021import org.apache.commons.logging.Log; 022import org.apache.commons.logging.LogFactory; 023import org.nuxeo.lib.stream.computation.Watermark; 024 025/** 026 * Keep track of minimum and maximum watermark level. On checkpoint move the low watermark to the previous maximum mark. 027 * 028 * @since 9.3 029 */ 030public class WatermarkMonotonicInterval { 031 private static final Log log = LogFactory.getLog(WatermarkMonotonicInterval.class); 032 033 protected volatile Watermark low = Watermark.LOWEST; 034 035 protected Watermark lowest = Watermark.LOWEST; 036 037 protected Watermark high = Watermark.LOWEST; 038 039 /** 040 * Take in account the watermark. 041 * 042 * @implNote Not thread safe. 043 */ 044 public long mark(long watermarkValue) { 045 return mark(Watermark.ofValue(watermarkValue)); 046 } 047 048 /** 049 * Take in account the watermark. 050 * 051 * @implNote Not thread safe. 052 */ 053 public long mark(Watermark watermark) { 054 if (Watermark.LOWEST.equals(low)) { 055 low = high = watermark; 056 } else if (watermark.compareTo(low) < 0) { 057 if (watermark.compareTo(lowest) < 0) { 058 // low watermark must increase to be monotonic 059 if (log.isTraceEnabled()) { 060 log.trace("receive too low watermark, rejected " + watermark + " lowest: " + lowest); 061 } 062 low = lowest; 063 } else { 064 low = watermark; 065 } 066 } 067 if (watermark.compareTo(high) > 0) { 068 high = watermark; 069 } 070 return low.getValue(); 071 } 072 073 /** 074 * Move the low watermark to the highest mark. Returns the low watermark that should be monotonic (the value 075 * returned here never decrease). 076 * 077 * @implNote Not thread safe. 078 */ 079 public long checkpoint() { 080 low = Watermark.completedOf(high); 081 lowest = low; 082 return low.getValue(); 083 } 084 085 public boolean isDone(long timestamp) { 086 return low.isDone(timestamp); 087 } 088 089 /** 090 * Returns the low mark. The value can decrease but not under the last checkpoint value. 091 * 092 * @implNote Thread safe usage. 093 */ 094 public Watermark getLow() { 095 return low; 096 } 097 098 public Watermark getHigh() { 099 return high; 100 } 101 102 @Override 103 public String toString() { 104 return "WatermarkInterval{" + "low=" + low + ", high=" + high + '}'; 105 } 106 107}