diff --git a/core/src/main/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBolt.java b/core/src/main/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBolt.java index 2fb09e36e..137035b76 100644 --- a/core/src/main/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBolt.java +++ b/core/src/main/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBolt.java @@ -21,6 +21,7 @@ import com.github.benmanes.caffeine.cache.Caffeine; import java.time.Instant; import java.time.format.DateTimeFormatter; +import java.time.format.DateTimeParseException; import java.util.Calendar; import java.util.Date; import java.util.Map; @@ -173,14 +174,26 @@ public void execute(Tuple tuple) { // changing the status or scheduling. String dateInMetadata = metadata.getFirstValue(AS_IS_NEXTFETCHDATE_METADATA); if (dateInMetadata != null) { - Date nextFetch = Date.from(Instant.parse(dateInMetadata)); + Date nextFetch = null; try { - store(url, status, mdTransfer.filter(metadata), Optional.of(nextFetch), tuple); - return; - } catch (Exception e) { - LOG.error("Exception caught when storing", e); - collector.fail(tuple); - return; + nextFetch = Date.from(Instant.parse(dateInMetadata)); + } catch (DateTimeParseException | IllegalArgumentException e) { + // not a date we can use: schedule the URL normally + LOG.error( + "Invalid value {} for {} on {}", + dateInMetadata, + AS_IS_NEXTFETCHDATE_METADATA, + url); + } + if (nextFetch != null) { + try { + store(url, status, mdTransfer.filter(metadata), Optional.of(nextFetch), tuple); + return; + } catch (Exception e) { + LOG.error("Exception caught when storing", e); + collector.fail(tuple); + return; + } } } diff --git a/core/src/main/java/org/apache/stormcrawler/util/StringTabScheme.java b/core/src/main/java/org/apache/stormcrawler/util/StringTabScheme.java index b0a2ce20e..0d5bccb33 100644 --- a/core/src/main/java/org/apache/stormcrawler/util/StringTabScheme.java +++ b/core/src/main/java/org/apache/stormcrawler/util/StringTabScheme.java @@ -25,7 +25,14 @@ import org.apache.storm.tuple.Values; import org.apache.stormcrawler.Metadata; -/** Converts a byte array into URL + metadata. */ +/** + * Converts a byte array into URL + metadata. + * + *

The format has no escaping: the keys and values are separated by tabulations and a key ends at + * the first equal sign. A value containing a tabulation is therefore read back as several key / + * value pairs, so this scheme suits seed files and other input produced with that limitation in + * mind. + */ public class StringTabScheme implements Scheme { @Override diff --git a/core/src/test/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBoltTest.java b/core/src/test/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBoltTest.java new file mode 100644 index 000000000..d901afb98 --- /dev/null +++ b/core/src/test/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBoltTest.java @@ -0,0 +1,122 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.persistence; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Optional; +import org.apache.storm.task.OutputCollector; +import org.apache.storm.tuple.Tuple; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.TestUtil; +import org.apache.stormcrawler.util.MetadataTransfer; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** + * The date passed in {@link AbstractStatusUpdaterBolt#AS_IS_NEXTFETCHDATE_METADATA} comes from the + * metadata and can be anything, so it must be parsed defensively. + */ +class AbstractStatusUpdaterBoltTest { + + private static final String URL = "http://example.com/"; + + /** Records what the bolt asked to store. */ + private static class RecordingStatusUpdaterBolt extends AbstractStatusUpdaterBolt { + + Optional nextFetch; + int stored = 0; + + @Override + public void store( + String url, Status status, Metadata metadata, Optional nextFetch, Tuple t) { + this.nextFetch = nextFetch; + this.stored++; + ack(t, url); + } + } + + private RecordingStatusUpdaterBolt bolt; + + private OutputCollector collector; + + @BeforeEach + void setUp() { + collector = mock(OutputCollector.class); + bolt = new RecordingStatusUpdaterBolt(); + Map conf = new HashMap<>(); + conf.put(AbstractStatusUpdaterBolt.useCacheParamName, Boolean.FALSE); + conf.put(Scheduler.schedulerClassParamName, DefaultScheduler.class.getName()); + conf.put(MetadataTransfer.metadataTransferClassParamName, MetadataTransfer.class.getName()); + bolt.prepare(conf, TestUtil.getMockedTopologyContext(), collector); + } + + private static Tuple statusTuple(Metadata metadata) { + Tuple tuple = mock(Tuple.class); + when(tuple.getStringByField("url")).thenReturn(URL); + when(tuple.getValueByField("metadata")).thenReturn(metadata); + when(tuple.getValueByField("status")).thenReturn(Status.FETCHED); + return tuple; + } + + @Test + void validNextFetchDateIsUsedAsIs() { + Metadata metadata = new Metadata(); + metadata.setValue( + AbstractStatusUpdaterBolt.AS_IS_NEXTFETCHDATE_METADATA, "2026-01-02T03:04:05Z"); + + bolt.execute(statusTuple(metadata)); + + assertEquals(1, bolt.stored); + assertNotNull(bolt.nextFetch); + assertTrue(bolt.nextFetch.isPresent()); + assertEquals("2026-01-02T03:04:05Z", bolt.nextFetch.get().toInstant().toString()); + } + + @Test + void invalidNextFetchDateIsIgnoredAndTheUrlIsScheduled() { + Metadata metadata = new Metadata(); + metadata.setValue(AbstractStatusUpdaterBolt.AS_IS_NEXTFETCHDATE_METADATA, "NOT-A-DATE"); + + bolt.execute(statusTuple(metadata)); + + assertEquals(1, bolt.stored, "the URL must still be stored"); + assertNotNull(bolt.nextFetch); + } + + @Test + void outOfRangeNextFetchDateIsIgnoredAndTheUrlIsScheduled() { + Metadata metadata = new Metadata(); + // parses as an instant but does not fit a java.util.Date + metadata.setValue( + AbstractStatusUpdaterBolt.AS_IS_NEXTFETCHDATE_METADATA, + "+1000000000-12-31T23:59:59Z"); + + bolt.execute(statusTuple(metadata)); + + assertEquals(1, bolt.stored, "the URL must still be stored"); + assertNotNull(bolt.nextFetch); + } +} diff --git a/external/sql/src/main/java/org/apache/stormcrawler/sql/MetadataColumn.java b/external/sql/src/main/java/org/apache/stormcrawler/sql/MetadataColumn.java new file mode 100644 index 000000000..5bd6c8332 --- /dev/null +++ b/external/sql/src/main/java/org/apache/stormcrawler/sql/MetadataColumn.java @@ -0,0 +1,156 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.sql; + +import org.apache.stormcrawler.Metadata; + +/** + * Encodes and decodes the content of the metadata column of the status table. + * + *

The keys and values are held in a single column as a list of key=value pairs + * separated by tabulations. Since a value can contain any character, including the separators + * themselves, the pairs written by {@link #encode(Metadata)} have the backslash, tabulation, + * carriage return, line feed and equal sign characters escaped, so that the decoding is + * unambiguous: a value can never introduce a key of its own. + * + *

An escaped column is made of {@link #FORMAT_MARKER} followed by the pairs, each preceded by a + * tabulation. Columns written before the escaping was introduced start with a tabulation or are + * empty; they are decoded verbatim, as they were previously, so that an existing table keeps + * working. + */ +final class MetadataColumn { + + /** Marks a column whose keys and values are escaped. */ + static final String FORMAT_MARKER = "v1"; + + private static final char SEPARATOR = '\t'; + + private static final char KEY_VALUE_SEPARATOR = '='; + + private static final char ESCAPE = '\\'; + + private MetadataColumn() {} + + /** Returns the representation of the metadata to store in the column. */ + static String encode(Metadata metadata) { + final StringBuilder column = new StringBuilder(FORMAT_MARKER); + for (String key : metadata.keySet()) { + // a key holding no readable value, such as the empty key a column written before the + // escaping can carry, is not written out + final String[] values = metadata.getValues(key); + if (values == null) { + continue; + } + for (String value : values) { + if (value == null) { + continue; + } + column.append(SEPARATOR) + .append(escape(key)) + .append(KEY_VALUE_SEPARATOR) + .append(escape(value)); + } + } + return column.toString(); + } + + /** Rebuilds the metadata from the content of the column, which can be null or empty. */ + static Metadata decode(String column) { + final Metadata metadata = new Metadata(); + if (column == null || column.isEmpty()) { + return metadata; + } + // the marker is only one when it is on its own or followed by a separator, so that a + // column hand written without the leading tabulation is not mistaken for an escaped one + final boolean escaped = + column.equals(FORMAT_MARKER) || column.startsWith(FORMAT_MARKER + SEPARATOR); + final String pairs = escaped ? column.substring(FORMAT_MARKER.length()) : column; + for (String pair : pairs.split("\t")) { + if (pair.isEmpty()) { + continue; + } + final int separator = + escaped ? separatorIndex(pair) : pair.indexOf(KEY_VALUE_SEPARATOR); + if (separator == -1) { + continue; + } + final String key = pair.substring(0, separator); + final String value = pair.substring(separator + 1); + if (escaped) { + metadata.addValue(unescape(key), unescape(value)); + } else { + metadata.addValue(key, value); + } + } + return metadata; + } + + /** Position of the first key / value separator which is not escaped, -1 if there is none. */ + private static int separatorIndex(String pair) { + boolean escaping = false; + for (int i = 0; i < pair.length(); i++) { + final char c = pair.charAt(i); + if (escaping) { + escaping = false; + } else if (c == ESCAPE) { + escaping = true; + } else if (c == KEY_VALUE_SEPARATOR) { + return i; + } + } + return -1; + } + + private static String escape(String value) { + final StringBuilder escaped = new StringBuilder(value.length()); + for (int i = 0; i < value.length(); i++) { + final char c = value.charAt(i); + switch (c) { + case ESCAPE -> escaped.append("\\\\"); + case SEPARATOR -> escaped.append("\\t"); + case '\n' -> escaped.append("\\n"); + case '\r' -> escaped.append("\\r"); + case KEY_VALUE_SEPARATOR -> escaped.append("\\="); + default -> escaped.append(c); + } + } + return escaped.toString(); + } + + private static String unescape(String value) { + final StringBuilder unescaped = new StringBuilder(value.length()); + for (int i = 0; i < value.length(); i++) { + final char c = value.charAt(i); + if (c != ESCAPE || i == value.length() - 1) { + unescaped.append(c); + continue; + } + final char next = value.charAt(++i); + switch (next) { + case '\\' -> unescaped.append(ESCAPE); + case 't' -> unescaped.append(SEPARATOR); + case 'n' -> unescaped.append('\n'); + case 'r' -> unescaped.append('\r'); + case '=' -> unescaped.append(KEY_VALUE_SEPARATOR); + // not a sequence we produce: keep it as it is + default -> unescaped.append(c).append(next); + } + } + return unescaped.toString(); + } +} diff --git a/external/sql/src/main/java/org/apache/stormcrawler/sql/SQLSpout.java b/external/sql/src/main/java/org/apache/stormcrawler/sql/SQLSpout.java index 0129cf3dd..cc85d45a0 100644 --- a/external/sql/src/main/java/org/apache/stormcrawler/sql/SQLSpout.java +++ b/external/sql/src/main/java/org/apache/stormcrawler/sql/SQLSpout.java @@ -17,25 +17,20 @@ package org.apache.stormcrawler.sql; -import java.nio.ByteBuffer; -import java.nio.charset.StandardCharsets; import java.sql.Connection; import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; import java.time.Instant; -import java.util.List; import java.util.Locale; import java.util.Map; -import org.apache.storm.spout.Scheme; import org.apache.storm.spout.SpoutOutputCollector; import org.apache.storm.task.TopologyContext; import org.apache.storm.topology.OutputFieldsDeclarer; -import org.apache.stormcrawler.Metadata; +import org.apache.storm.tuple.Fields; import org.apache.stormcrawler.persistence.AbstractQueryingSpout; import org.apache.stormcrawler.util.ConfUtils; -import org.apache.stormcrawler.util.StringTabScheme; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,8 +38,6 @@ public class SQLSpout extends AbstractQueryingSpout { public static final Logger LOG = LoggerFactory.getLogger(SQLSpout.class); - private static final Scheme SCHEME = new StringTabScheme(); - private static final String BASE_SQL = """ SELECT * @@ -137,7 +130,7 @@ public void open( @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { - declarer.declare(SCHEME.getOutputFields()); + declarer.declare(new Fields("url", "metadata")); } @Override @@ -235,14 +228,7 @@ private int processRow(final ResultSet rs) throws SQLException { return 1; } - final String normalisedMetadata = - (metadata == null || metadata.startsWith("\t")) ? metadata : "\t" + metadata; - - final String urlWithMetadata = String.format(Locale.ROOT, "%s%s", url, normalisedMetadata); - final List v = - SCHEME.deserialize( - ByteBuffer.wrap(urlWithMetadata.getBytes(StandardCharsets.UTF_8))); - buffer.add(url, (Metadata) v.get(1)); + buffer.add(url, MetadataColumn.decode(metadata)); return 0; } diff --git a/external/sql/src/main/java/org/apache/stormcrawler/sql/StatusUpdaterBolt.java b/external/sql/src/main/java/org/apache/stormcrawler/sql/StatusUpdaterBolt.java index 80642dba0..fe120a4b7 100644 --- a/external/sql/src/main/java/org/apache/stormcrawler/sql/StatusUpdaterBolt.java +++ b/external/sql/src/main/java/org/apache/stormcrawler/sql/StatusUpdaterBolt.java @@ -155,13 +155,7 @@ public synchronized void store( return; } - final StringBuilder mdAsString = new StringBuilder(); - for (String mdKey : metadata.keySet()) { - String[] vals = metadata.getValues(mdKey); - for (String v : vals) { - mdAsString.append("\t").append(mdKey).append("=").append(v); - } - } + final String mdAsString = MetadataColumn.encode(metadata); int partition = 0; String partitionKey = partitioner.getPartition(url, metadata); @@ -213,7 +207,7 @@ private void populate( final String url, final Status status, final Optional nextFetch, - final StringBuilder mdAsString, + final String mdAsString, final int partition, final String partitionKey, final PreparedStatement preparedStmt) @@ -227,7 +221,7 @@ private void populate( // a value so large it means it will never be refetched preparedStmt.setObject(3, NEVER); } - preparedStmt.setString(4, mdAsString.toString()); + preparedStmt.setString(4, mdAsString); preparedStmt.setInt(5, partition); preparedStmt.setString(6, partitionKey); } diff --git a/external/sql/src/test/java/org/apache/stormcrawler/sql/StatusMetadataRoundTripTest.java b/external/sql/src/test/java/org/apache/stormcrawler/sql/StatusMetadataRoundTripTest.java new file mode 100644 index 000000000..eb1b246c3 --- /dev/null +++ b/external/sql/src/test/java/org/apache/stormcrawler/sql/StatusMetadataRoundTripTest.java @@ -0,0 +1,169 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to you under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.stormcrawler.sql; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; + +import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Method; +import java.lang.reflect.Proxy; +import java.sql.PreparedStatement; +import java.util.HashMap; +import java.util.Optional; +import org.apache.storm.task.OutputCollector; +import org.apache.stormcrawler.Metadata; +import org.apache.stormcrawler.TestOutputCollector; +import org.apache.stormcrawler.metrics.ScopedCounter; +import org.apache.stormcrawler.persistence.Status; +import org.apache.stormcrawler.util.URLPartitioner; +import org.junit.jupiter.api.Test; + +/** + * The metadata column written by {@link StatusUpdaterBolt} is read back by {@link SQLSpout}. + * Whatever goes in must come out unchanged, and a value must never introduce a key of its own. + * Needs no database: the prepared statement is a recording stub. + */ +class StatusMetadataRoundTripTest { + + private static final String URL = "http://example.com/"; + + /** Captures the value bound to the metadata column. */ + private static class MetadataCapture implements InvocationHandler { + String metadataColumn; + + @Override + public Object invoke(Object proxy, Method method, Object[] args) { + if ("setString".equals(method.getName()) && Integer.valueOf(4).equals(args[0])) { + metadataColumn = (String) args[1]; + } + if ("executeUpdate".equals(method.getName())) { + return Integer.valueOf(1); + } + return null; + } + } + + private static void set(Object target, Class owner, String name, Object value) + throws Exception { + Field f = owner.getDeclaredField(name); + f.setAccessible(true); + f.set(target, value); + } + + /** Runs store() against a recording statement and returns the metadata column it wrote. */ + private static String storedMetadataColumn(Metadata metadata) throws Exception { + StatusUpdaterBolt bolt = new StatusUpdaterBolt(); + URLPartitioner partitioner = new URLPartitioner(); + partitioner.configure(new HashMap<>()); + MetadataCapture capture = new MetadataCapture(); + PreparedStatement statement = + (PreparedStatement) + Proxy.newProxyInstance( + StatusMetadataRoundTripTest.class.getClassLoader(), + new Class[] {PreparedStatement.class}, + capture); + ScopedCounter counter = scope -> incrementBy -> {}; + set(bolt, StatusUpdaterBolt.class, "partitioner", partitioner); + set(bolt, StatusUpdaterBolt.class, "updatePreparedStmt", statement); + set(bolt, StatusUpdaterBolt.class, "eventCounter", counter); + set(bolt, bolt.getClass().getSuperclass(), "useCache", Boolean.FALSE); + set( + bolt, + bolt.getClass().getSuperclass(), + "collector", + new OutputCollector(new TestOutputCollector())); + bolt.store(URL, Status.FETCHED, metadata, Optional.empty(), null); + return capture.metadataColumn; + } + + /** Reads the column back the way SQLSpout does. */ + private static Metadata readBack(String metadataColumn) { + return MetadataColumn.decode(metadataColumn); + } + + @Test + void tabInAValueDoesNotBecomeAKey() throws Exception { + Metadata metadata = new Metadata(); + // _redirTo is persisted by default and holds the raw Location header + metadata.setValue("_redirTo", "/a\tstatus.store.as.is.with.nextfetchdate=NOT-A-DATE"); + Metadata back = readBack(storedMetadataColumn(metadata)); + assertNull( + back.getFirstValue("status.store.as.is.with.nextfetchdate"), + "a stored value must not introduce a key when it is read back"); + assertEquals( + metadata.getFirstValue("_redirTo"), + back.getFirstValue("_redirTo"), + "the stored value must come back unchanged"); + } + + @Test + void tabbedValueDoesNotMintDepthKeys() throws Exception { + Metadata metadata = new Metadata(); + metadata.setValue("_redirTo", "/a\tdepth=0\tmax.depth=99999"); + Metadata back = readBack(storedMetadataColumn(metadata)); + assertNull(back.getFirstValue("depth"), "depth must not be settable from a stored value"); + assertNull( + back.getFirstValue("max.depth"), + "max.depth must not be settable from a stored value"); + } + + @Test + void separatorsAndBackslashesInValuesSurviveTheRoundTrip() throws Exception { + Metadata metadata = new Metadata(); + metadata.setValue("_redirTo", "/a?b=c\td\\e\nf\rg\\t"); + metadata.setValues("multi", new String[] {"first=1", "second\ttoo"}); + Metadata back = readBack(storedMetadataColumn(metadata)); + assertEquals("/a?b=c\td\\e\nf\rg\\t", back.getFirstValue("_redirTo")); + assertEquals("first=1", back.getValues("multi")[0]); + assertEquals("second\ttoo", back.getValues("multi")[1]); + } + + @Test + void columnWrittenBeforeTheEscapingIsStillRead() { + Metadata back = readBack("\tdepth=1\t_redirTo=http://example.com/a"); + assertEquals("1", back.getFirstValue("depth")); + assertEquals("http://example.com/a", back.getFirstValue("_redirTo")); + } + + @Test + void columnWrittenBeforeTheEscapingWithoutALeadingTabulationIsStillRead() { + Metadata back = readBack("v1key=value\tdepth=1"); + assertEquals("value", back.getFirstValue("v1key"), "the marker is not a key prefix"); + assertEquals("1", back.getFirstValue("depth")); + } + + @Test + void columnWrittenBeforeTheEscapingWithAnEmptyKeyIsRewrittenWithoutIt() throws Exception { + Metadata back = readBack("\t=x\tdepth=1"); + assertEquals("1", back.getFirstValue("depth")); + assertNull(back.getFirstValue(""), "an empty key cannot be read back from a metadata"); + assertEquals( + "v1\tdepth=1", + storedMetadataColumn(back), + "a key which cannot be read back is not written out again"); + } + + @Test + void emptyColumnGivesEmptyMetadata() { + assertEquals(0, readBack(null).size()); + assertEquals(0, readBack("").size()); + assertEquals(0, readBack(MetadataColumn.encode(new Metadata())).size()); + } +}