Escape the metadata column of the SQL status store

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;
+                }
             }
         }
 

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.
+ *
+ * <p>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

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<Date> nextFetch;
+        int stored = 0;
+
+        @Override
+        public void store(
+                String url, Status status, Metadata metadata, Optional<Date> 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<String, Object> 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);
+    }
+}

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 <code>metadata</code> column of the status table.
+ *
+ * <p>The keys and values are held in a single column as a list of <code>key=value</code> 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.
+ *
+ * <p>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();
+    }
+}

external/sql/src/main/java/org/apache/stormcrawler/sql/SQLSpout.java

@@ -17,34 +17,27 @@
 
 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;
 
 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<Object> 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;
     }