From a9365f05f5e620959b10296d26aeb72aafed26e8 Mon Sep 17 00:00:00 2001 From: Richard Zowalla <13417392+rzo1@users.noreply.github.com> Date: Thu, 27 Aug 2026 14:20:10 +0200 Subject: [PATCH] Escape the metadata column of the SQL status store The SQL status store flattened the metadata into one column as tab separated key=value pairs without escaping, so a value containing a tabulation came back from SQLSpout as extra keys and any legitimate value with a tabulation was corrupted. Keys and values are now escaped when the column is written and unescaped when it is read, which makes the round trip lossless rather than dropping characters silently. Columns written by the previous code start with a tabulation, columns written now start with a format marker, so existing rows keep being decoded exactly as before. Rows written from now on are not readable by earlier versions, which see the marker as an extra key. SQLSpout no longer decodes the column with StringTabScheme, whose format cannot represent a tabulation in a value. AbstractStatusUpdaterBolt also parsed status.store.as.is.with.nextfetchdate outside its try/catch, so a stored value that is not a valid instant threw out of execute. It is now logged and the URL scheduled normally, including a value which parses as an instant but does not fit a java.util.Date. Encoding a metadata no longer fails on a key which holds no readable value, such as the empty key a column written before the escaping can produce. --- .../AbstractStatusUpdaterBolt.java | 27 ++- .../stormcrawler/util/StringTabScheme.java | 9 +- .../AbstractStatusUpdaterBoltTest.java | 122 +++++++++++++ .../stormcrawler/sql/MetadataColumn.java | 156 ++++++++++++++++ .../org/apache/stormcrawler/sql/SQLSpout.java | 20 +-- .../stormcrawler/sql/StatusUpdaterBolt.java | 12 +- .../sql/StatusMetadataRoundTripTest.java | 169 ++++++++++++++++++ 7 files changed, 481 insertions(+), 34 deletions(-) create mode 100644 core/src/test/java/org/apache/stormcrawler/persistence/AbstractStatusUpdaterBoltTest.java create mode 100644 external/sql/src/main/java/org/apache/stormcrawler/sql/MetadataColumn.java create mode 100644 external/sql/src/test/java/org/apache/stormcrawler/sql/StatusMetadataRoundTripTest.java 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 The keys and values are held in a single column as a list of 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 Listmetadata column of the status table.
+ *
+ * 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.
+ *
+ *