Smalyshev has uploaded a new change for review. (
https://gerrit.wikimedia.org/r/330822 )
Change subject: Do not try to process same change ID twice
......................................................................
Do not try to process same change ID twice
Bug: T153932
Change-Id: I5c970930b0da99f6567279ce0ebfbba040bd7f91
---
M tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
M
tools/src/main/java/org/wikidata/query/rdf/tool/change/RecentChangesPoller.java
M
tools/src/test/java/org/wikidata/query/rdf/tool/change/RecentChangesPollerUnitTest.java
3 files changed, 46 insertions(+), 29 deletions(-)
git pull ssh://gerrit.wikimedia.org:29418/wikidata/query/rdf
refs/changes/22/330822/1
diff --git a/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
index 611dc55..b93a952 100644
--- a/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
+++ b/tools/src/main/java/org/wikidata/query/rdf/tool/Update.java
@@ -315,6 +315,7 @@
}
} while (batch == null);
log.debug("{} changes in batch", batch.changes().size());
+ Date oldDate = null;
while (true) {
try {
handleChanges(batch);
@@ -326,7 +327,11 @@
* have some updates.
*/
leftOffDate = new Date(batch.leftOffDate().getTime() -
SECONDS.toMillis(1));
- rdfRepository.updateLeftOffTime(leftOffDate);
+ if (oldDate == null || !oldDate.equals(leftOffDate)) {
+ // Do not update repo with the same date
+ oldDate = leftOffDate;
+ rdfRepository.updateLeftOffTime(leftOffDate);
+ }
}
// TODO wrap all retry-able exceptions in a special exception
batchAdvanced.mark(batch.advanced());
@@ -435,10 +440,11 @@
* @throws InterruptedException if the process was interrupted while
waiting
* during the pollDelay or waiting on something else
*/
- private B nextBatch(B batch) throws InterruptedException {
+ private B nextBatch(B prevBatch) throws InterruptedException {
+ B batch;
while (true) {
try {
- batch = changeSource.nextBatch(batch);
+ batch = changeSource.nextBatch(prevBatch);
} catch (RetryableException e) {
log.warn("Retryable error fetching next batch. Retrying.", e);
continue;
diff --git
a/tools/src/main/java/org/wikidata/query/rdf/tool/change/RecentChangesPoller.java
b/tools/src/main/java/org/wikidata/query/rdf/tool/change/RecentChangesPoller.java
index 489ef9c..09ca176 100644
---
a/tools/src/main/java/org/wikidata/query/rdf/tool/change/RecentChangesPoller.java
+++
b/tools/src/main/java/org/wikidata/query/rdf/tool/change/RecentChangesPoller.java
@@ -4,8 +4,10 @@
import java.text.DateFormat;
import java.util.Date;
+import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Set;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
@@ -51,7 +53,7 @@
@Override
public Batch nextBatch(Batch lastBatch) throws RetryableException {
- return batch(lastBatch.leftOffDate, lastBatch.nextContinue);
+ return batch(lastBatch.leftOffDate, lastBatch);
}
/**
@@ -62,20 +64,21 @@
* The date where we last left off.
*/
private final Date leftOffDate;
+
/**
- * The continue object that must be sent to wikibase to continue where
- * we left off.
+ * The set of the rcid's this batch seen.
+ * Note that some IDs may be seen but not processed
+ * due to duplicates, etc.
*/
- private final JSONObject nextContinue;
+ private final Set<Long> seenIDs;
/**
* A batch that will next continue using the continue parameter.
*/
- private Batch(ImmutableList<Change> changes, long advanced, String
leftOff, Date nextStartTime,
- JSONObject nextContinue) {
+ private Batch(ImmutableList<Change> changes, long advanced, String
leftOff, Date nextStartTime, Set<Long> seenIDs) {
super(changes, advanced, leftOff);
leftOffDate = nextStartTime;
- this.nextContinue = nextContinue;
+ this.seenIDs = seenIDs;
}
@Override
@@ -90,8 +93,16 @@
@Override
public String leftOffHuman() {
- return leftOffDate.toString() + " (next: " +
nextContinue.get("rccontinue").toString()
- + ")";
+ return WikibaseRepository.inputDateFormat().format(leftOffDate);
+ // + " (next: " + nextContinue.get("rccontinue").toString() + ")";
+ }
+
+ /**
+ * Get the list of IDs this batch has seen.
+ * @return
+ */
+ public Set<Long> getSeenIDs() {
+ return seenIDs;
}
}
@@ -101,10 +112,9 @@
* @throws RetryableException on parse failure
*/
@SuppressWarnings("checkstyle:cyclomaticcomplexity")
- private Batch batch(Date lastNextStartTime, JSONObject lastNextContinue)
throws RetryableException {
+ private Batch batch(Date lastNextStartTime, Batch lastBatch) throws
RetryableException {
try {
@SuppressWarnings("unchecked")
- final Change continueChange =
wikibase.getChangeFromContinue((Map<String, Object>)lastNextContinue);
JSONObject recentChanges =
wikibase.fetchRecentChangesBackoff(lastNextStartTime, batchSize, true);
// Using LinkedHashMap here so that changes came out sorted by
order of arrival
Map<String, Change> changesByTitle = new LinkedHashMap<>();
@@ -112,6 +122,8 @@
long nextStartTime = lastNextStartTime.getTime();
JSONArray result = (JSONArray) ((JSONObject)
recentChanges.get("query")).get("recentchanges");
DateFormat df = inputDateFormat();
+ Set<Long> seenIds = new HashSet<>();
+
for (Object rco : result) {
JSONObject rc = (JSONObject) rco;
long namespace = (long) rc.get("ns");
@@ -121,7 +133,13 @@
continue;
}
if (!wikibase.isValidEntity(rc.get("title").toString())) {
- log.info("Skipping change with bogus id: {}",
rc.get("title").toString());
+ log.info("Skipping change with bogus title: {}",
rc.get("title").toString());
+ continue;
+ }
+ seenIds.add(rcid);
+ if (lastBatch != null &&
lastBatch.getSeenIDs().contains(rcid)) {
+ // This change was in the last batch
+ log.debug("Skipping repeated change with rcid {}", rcid);
continue;
}
// Looks like we can not rely on changes appearing in order, so we have to
take them all and let SPARQL
@@ -152,19 +170,11 @@
nextStartTime = Math.max(nextStartTime, timestamp.getTime());
}
ImmutableList<Change> changes =
ImmutableList.copyOf(changesByTitle.values());
- if (nextContinue == null) {
- if (changes.size() != 0) {
- // Create fake rccontinue so we continue from last known
change
- nextContinue =
wikibase.getContinueObject(changes.get(changes.size() - 1));
- } else {
- if (result.size() >= batchSize) {
- // We have a problem here - due to backoff, we did not
fetch any new items
- // Try to advance one second, even though we risk to
lose a change
- log.info("Backoff overflow, advancing one second");
- nextStartTime += 1000;
- }
- nextContinue = lastNextContinue;
- }
+ if (nextContinue == null && changes.size() == 0 && result.size()
>= batchSize) {
+ // We have a problem here - due to backoff, we did not fetch
any new items
+ // Try to advance one second, even though we risk to lose a
change
+ log.info("Backoff overflow, advancing one second");
+ nextStartTime += 1000;
}
if (changes.size() != 0) {
@@ -179,7 +189,7 @@
// be sure we got the whole second
String upTo = inputDateFormat().format(new Date(nextStartTime -
1000));
long advanced = nextStartTime - lastNextStartTime.getTime();
- return new Batch(changes, advanced, upTo, new Date(nextStartTime),
nextContinue);
+ return new Batch(changes, advanced, upTo, new Date(nextStartTime),
seenIds);
} catch (java.text.ParseException e) {
throw new RetryableException("Parse error from api", e);
}
diff --git
a/tools/src/test/java/org/wikidata/query/rdf/tool/change/RecentChangesPollerUnitTest.java
b/tools/src/test/java/org/wikidata/query/rdf/tool/change/RecentChangesPollerUnitTest.java
index 170e1c3..8d19629 100644
---
a/tools/src/test/java/org/wikidata/query/rdf/tool/change/RecentChangesPollerUnitTest.java
+++
b/tools/src/test/java/org/wikidata/query/rdf/tool/change/RecentChangesPollerUnitTest.java
@@ -122,6 +122,7 @@
Batch batch = poller.firstBatch();
assertThat(batch.changes(), hasSize(2));
assertEquals(7, batch.changes().get(1).rcid());
+ assertEquals(date, batch.leftOffHuman());
ArgumentCaptor<Date> argument = ArgumentCaptor.forClass(Date.class);
--
To view, visit https://gerrit.wikimedia.org/r/330822
To unsubscribe, visit https://gerrit.wikimedia.org/r/settings
Gerrit-MessageType: newchange
Gerrit-Change-Id: I5c970930b0da99f6567279ce0ebfbba040bd7f91
Gerrit-PatchSet: 1
Gerrit-Project: wikidata/query/rdf
Gerrit-Branch: master
Gerrit-Owner: Smalyshev <[email protected]>
_______________________________________________
MediaWiki-commits mailing list
[email protected]
https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits
