This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/stormcrawler.git
The following commit(s) were added to refs/heads/main by this push:
new f520b4e4 Escape the metadata column of the SQL status store (#2076)
f520b4e4 is described below
commit f520b4e4d3ef15df84eff84d80921c327768d7e7
Author: Richard Zowalla <[email protected]>
AuthorDate: Sat Aug 29 13:22:10 2026 +0200
Escape the metadata column of the SQL status store (#2076)
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.
---
.../persistence/AbstractStatusUpdaterBolt.java | 27 +++-
.../apache/stormcrawler/util/StringTabScheme.java | 9 +-
.../persistence/AbstractStatusUpdaterBoltTest.java | 122 +++++++++++++++
.../apache/stormcrawler/sql/MetadataColumn.java | 156 +++++++++++++++++++
.../java/org/apache/stormcrawler/sql/SQLSpout.java | 20 +--
.../apache/stormcrawler/sql/StatusUpdaterBolt.java | 12 +-
.../sql/StatusMetadataRoundTripTest.java | 169 +++++++++++++++++++++
7 files changed, 481 insertions(+), 34 deletions(-)
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 2fb09e36..137035b7 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.Cache;
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 abstract class AbstractStatusUpdaterBolt extends
BaseRichBolt {
// 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 b0a2ce20..0d5bccb3 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.Fields;
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
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 00000000..d901afb9
--- /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<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);
+ }
+}
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 00000000..5bd6c833
--- /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 <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();
+ }
+}
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 0129cf3d..cc85d45a 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 class SQLSpout extends AbstractQueryingSpout {
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
- declarer.declare(SCHEME.getOutputFields());
+ declarer.declare(new Fields("url", "metadata"));
}
@Override
@@ -235,14 +228,7 @@ public class SQLSpout extends AbstractQueryingSpout {
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;
}
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 80642dba..fe120a4b 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 class StatusUpdaterBolt extends
AbstractStatusUpdaterBolt {
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 @@ public class StatusUpdaterBolt extends
AbstractStatusUpdaterBolt {
final String url,
final Status status,
final Optional<Date> nextFetch,
- final StringBuilder mdAsString,
+ final String mdAsString,
final int partition,
final String partitionKey,
final PreparedStatement preparedStmt)
@@ -227,7 +221,7 @@ public class StatusUpdaterBolt extends
AbstractStatusUpdaterBolt {
// 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 00000000..eb1b246c
--- /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());
+ }
+}
