Utils { public static void alignWithExternalRevisions(@NotNull NodeDocument rootDoc, @NotNull Clock clock, int clusterId) throws InterruptedException { Map<Integer, Revision> lastRevMap = checkNotNull(rootDoc).getLastRev(); long externalTime = Utils.getMaxExternalTimestamp(lastRevMap.values(), clusterId); long localTime = clock.getTime(); if (localTime < externalTime) { LOG.warn("Detected clock differences. Local time is '{}', " + "while most recent external time is '{}'. " + "Current _lastRev entries: {}", new Date(localTime), new Date(externalTime), lastRevMap.values()); double delay = ((double) externalTime - localTime) / 1000d; String fmt = "Background read will be delayed by %.1f seconds. " + "Please check system time on cluster nodes."; String msg = String.format(fmt, delay); LOG.warn(msg); while (localTime + 60000 < externalTime) { clock.waitUntil(localTime + 60000); localTime = clock.getTime(); delay = ((double) externalTime - localTime) / 1000d; LOG.warn(String.format(fmt, delay)); } clock.waitUntil(externalTime + 1); } else if (localTime == externalTime) { LOG.debug("Local and external time are equal. Waiting until local" + "time is more recent than external reported time."); clock.waitUntil(externalTime + 1); } }  static int pathDepth(String path); static int getIdDepth(Path path); @SuppressWarnings("unchecked") static int estimateMemoryUsage(Map<?, Object> map); static String escapePropertyName(String propertyName); static String unescapePropertyName(String key); static boolean isPropertyName(String key); static String getIdFromPath(@NotNull String path); static String getIdFromPath(@NotNull Path path); static StringBuilder encodeHexString(byte[] data, StringBuilder sb); @Nullable static String getParentId(String id); static boolean isLongPath(Path path); static boolean isIdFromLongPath(String id); static String getPathFromId(String id); static int getDepthFromId(String id); static Path getPreviousPathFor(Path path, Revision r, int height); static String getPreviousIdFor(Path path, Revision r, int height); static boolean isPreviousDocId(String id); static boolean isLeafPreviousDocId(String id); static void deepCopyMap(Map<K, Object> source, Map<K, Object> target); static String getKeyLowerLimit(Path path); static String getKeyUpperLimit(Path path); @Nullable static String getParentIdFromLowerLimit(String fromKey); static boolean isCommitted(@Nullable String tag); @NotNull static Revision resolveCommitRevision(@NotNull Revision rev,
                                                 @NotNull String tag); static void closeIfCloseable(Object obj); static String timestampToString(long timestamp); @Nullable static Revision max(@Nullable Revision a, @Nullable Revision b); @Nullable static Revision max(@Nullable Revision a,
                               @Nullable Revision b,
                               @NotNull Comparator<Revision> c); @Nullable static Revision min(@Nullable Revision a, @Nullable Revision b); @Nullable static Revision min(@Nullable Revision a,
                               @Nullable Revision b,
                               @NotNull Comparator<Revision> c); static Iterable<NodeDocument> getAllDocuments(final DocumentStore store); @NotNull static NodeDocument getRootDocument(@NotNull DocumentStore store); static Iterable<NodeDocument> getSelectedDocuments(
            DocumentStore store, String indexedProperty, long startValue, int batchSize); static Iterable<NodeDocument> getSelectedDocuments(
            DocumentStore store, String indexedProperty, long startValue); static boolean isHiddenPath(@NotNull String path); static Iterable<StringValue> asStringValueIterable(
            @NotNull Iterable<String> values); static Iterable<String> pathToId(@NotNull Iterable<String> paths); static long getMaxExternalTimestamp(Iterable<Revision> revisions,
                                               int localClusterId); static Long asLong(@Nullable Number n); @NotNull static RevisionVector getStartRevisions(@NotNull Iterable<ClusterNodeInfoDocument> clusterNodes); static long getMinTimestampForDiff(@NotNull RevisionVector fromRev,
                                              @NotNull RevisionVector toRev,
                                              @NotNull RevisionVector minRevisions); static boolean isGreaterOrEquals(@NotNull RevisionVector a,
                                            @NotNull RevisionVector b); static boolean isLocalChange(@NotNull RevisionVector from,
                                        @NotNull RevisionVector to,
                                        int clusterId); static CloseableIterable<T> abortingIterable(Iterable<T> iterable,
                                                            Predicate<T> p); static void alignWithExternalRevisions(@NotNull NodeDocument rootDoc,
                                                  @NotNull Clock clock,
                                                  int clusterId); static void joinQuietly(Thread... threads); static String getModuleVersion(); static void checkRevisionAge(DocumentStore store,
                                        ClusterNodeInfo info,
                                        Clock clock); static final int PATH_SHORT; static final int PATH_LONG; static final int NODE_NAME_LIMIT; static final Predicate<String> PROPERTY_OR_DELETED; }
@Test public void alignWithExternalRevisions() throws Exception { Clock c = new Clock.Virtual(); c.waitUntil(System.currentTimeMillis()); Revision lastRev1 = new Revision(c.getTime() - 1000, 0, 1); Revision lastRev2 = new Revision(c.getTime() + 1000, 0, 2); NodeDocument doc = new NodeDocument(new MemoryDocumentStore(), c.getTime()); UpdateOp op = new UpdateOp(Utils.getIdFromPath("/"), true); NodeDocument.setLastRev(op, lastRev1); NodeDocument.setLastRev(op, lastRev2); UpdateUtils.applyChanges(doc, op); Utils.alignWithExternalRevisions(doc, c, 2); assertThat(c.getTime(), is(lessThan(lastRev2.getTimestamp()))); Utils.alignWithExternalRevisions(doc, c, 1); assertThat(c.getTime(), is(greaterThan(lastRev2.getTimestamp()))); }