abhinav-phi commented on code in PR #2123:
URL: https://github.com/apache/stormcrawler/pull/2123#discussion_r3996892260
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -223,11 +254,19 @@ public void nextTuple() {
timeLastQuerySent = System.currentTimeMillis();
}
+ /** Checks the scheme of a stored URL against the configured {@code
protocols} list. */
+ protected boolean schemeAllowed(String url) {
+ int colon = url.indexOf(':');
+ if (colon <= 0) {
+ return false;
+ }
+ return allowedSchemes.contains(url.substring(0,
colon).toLowerCase(Locale.ROOT));
+ }
+
/**
* Returns the amount of time to wait if the backend was queried too
recently and needs
* throttling or -1 if the backend can be queried straight away.
- */
- private long throttleQueries() {
+ */ private long throttleQueries() {
Review Comment:
Fixed — signature back on its own line, and the head passes the rat job
(`mvn -B -Prat -DskipTests verify -Dskip.format.code=false`; green on 218d6cd9,
run 34651806805).
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -116,6 +122,21 @@ public void open(
buffer = URLBuffer.createInstance(stormConf);
+ /*
+ * The store is the crawl instruction set: whatever ends up in it is
+ * fetched. Schemes which are not configured for the crawl must not
+ * re-enter the topology from there, so rows are checked before they
+ * are emitted - URL filtering only runs on the discovery path.
+ */
+ allowedSchemes =
+ ConfUtils.loadListFromConf("protocols", stormConf).stream()
+ .map(String::trim)
+ .map(String::toLowerCase)
Review Comment:
Fixed — the scheme set is built with `Locale.ROOT` and the stored-URL
comparison lowercases with `Locale.ROOT` as well;
`AbstractQueryingSpoutSchemeTest.uppercaseSchemeIsAllowed` pins the
case-insensitive form.
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -191,7 +212,7 @@ public void nextTuple() {
timeLastQuerySent = System.currentTimeMillis();
}
- if (buffer.hasNext()) {
+ while (buffer.hasNext()) {
Review Comment:
Fixed — rejected rows no longer spin: they are emitted to the status stream
as `Status.ERROR` carrying the stored metadata and `STATUS_ERROR_CAUSE`, so the
updater marks them and they stop resurfacing on every query (0d955ce4,
acd35f2b). Wiring: every archetype `crawler.flux` now connects spout → status
updater (FIELDS grouping on url, same as the other status producers),
`SpoutStatusStreamWiringTest` asserts the edge for every archetype flux so it
cannot silently regress, and `internals.adoc` documents the upgrade for
hand-built topologies. `emitStatus` also logs once when the stream reaches no
task — the 3.x-flux-without-edge case @dpol1 flagged.
`urlsWithAnUnexpectedSchemeAreNotEmittedButReported` pins the ERROR emission.
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -201,11 +222,21 @@ public void nextTuple() {
}
List<Object> fields = buffer.next();
String url = fields.get(0).toString();
Review Comment:
Guarded in 0d955ce4 — a null from `buffer.next()` breaks the loop.
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -201,11 +222,21 @@ public void nextTuple() {
}
List<Object> fields = buffer.next();
String url = fields.get(0).toString();
+ if (!schemeAllowed(url)) {
+ LOG.warn(
+ "Stored URL {} not fetched: its scheme is not in the
configured list",
+ url);
+ eventCounter.scope("skipped.scheme").incrBy(1);
+ // try the next entry the buffer holds; a rejected row stays in
+ // the store and is skipped again on every query
+ continue;
+ }
this.collector.emit(fields, url);
beingProcessed.put(url, null);
eventCounter.scope("emitted").incrBy(1);
return;
- } else if (timestampEmptyBuffer == -1) {
+ }
+ if (timestampEmptyBuffer == -1 && !buffer.hasNext()) {
Review Comment:
Removed in 0d955ce4.
##########
core/src/main/java/org/apache/stormcrawler/persistence/AbstractQueryingSpout.java:
##########
@@ -223,6 +266,27 @@ public void nextTuple() {
timeLastQuerySent = System.currentTimeMillis();
}
+ /** Checks the scheme of a stored URL against the configured {@code
protocols} list. */
+ protected boolean schemeAllowed(String url) {
+ int colon = url.indexOf(':');
+ if (colon <= 0) {
+ return false;
+ }
+ return allowedSchemes.contains(url.substring(0,
colon).toLowerCase(Locale.ROOT));
+ }
+
+ /**
+ * Emits a tuple to the status stream so that the status updater processes
the status, e.g.
+ * removes a row whose URL the spout refuses to emit. The stored metadata
is passed on so that
+ * the status updater does not overwrite the row with an empty set.
+ */
+ protected void emitStatus(String url, Metadata metadata, Status status) {
+ if (metadata == null) {
+ metadata = new Metadata();
+ }
+ collector.emit(Constants.StatusStreamName, new Values(url, metadata,
status));
Review Comment:
Fixed in 218d6cd9 — the rejected-row ERROR now sets `STATUS_ERROR_CAUSE`
(`"scheme not in the configured protocols list"`), same convention as the
fetcher and the updater; pinned in
`urlsWithAnUnexpectedSchemeAreNotEmittedButReported`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
