This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 6d36a14d9 Require Kryo class registration in the state serializer
6d36a14d9 is described below
commit 6d36a14d9e871405dee3d1d0ac2db96af97dfb00
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Fri Aug 21 19:17:48 2026 +0200
Require Kryo class registration in the state serializer
Registration is keyed off topology.fall.back.on.java.serialization,
matching the tuple path in DefaultKryoFactory. Registers the types Storm
persists without the component declaring them (byte[], CheckPointState),
after the configured registrations so their ids are unaffected.
Clears the class resolver's name caches on reset: Kryo skips that when
registration is required, which would desynchronise a second read of
state written by an earlier release.
---
docs/State-checkpointing.md | 73 +++++--------
.../apache/storm/state/DefaultStateSerializer.java | 65 +++++++++++-
.../storm/state/DefaultStateSerializerTest.java | 117 +++++++++++++++++++++
3 files changed, 207 insertions(+), 48 deletions(-)
diff --git a/docs/State-checkpointing.md b/docs/State-checkpointing.md
index 687ea818f..e12074be4 100644
--- a/docs/State-checkpointing.md
+++ b/docs/State-checkpointing.md
@@ -211,6 +211,30 @@ The namespace is typically unique per task so that each
task can have its own st
State implementation should be available in the class path of Storm (by
placing them in the extlib directory).
+### Kryo class registration for state
+
+State values are serialized with Kryo before being written to an external
backend. When those bytes
+are read back, the serializer constructs only classes the topology has
registered, so what a worker
+builds from the stored bytes is bounded by its own configuration.
+
+The classes Storm itself persists are registered automatically. Register your
own key and value
+types in one of these ways:
+
+* `keyClass` / `valueClass` in `topology.state.provider.config` (see the
backend sections below), or
+* `topology.state.kryo.register`, a list of fully qualified class names.
+
+Types handled by Kryo out of the box, such as `String`, boxed primitives and
the common collections,
+need no registration. A value whose class is not registered fails with
+`IllegalArgumentException: Class is not registered`.
+
+Setting `topology.fall.back.on.java.serialization: true` restores the previous
behaviour, in which
+any class named in the stored bytes is constructed.
+
+**Upgrading with an existing state store.** State written before this
behaviour was introduced is
+still readable, including checkpoint state, as long as every type it contains
is registered. If a
+topology stored values of a custom class and does not register it, reads of
that state fail after
+the upgrade. Either register the class or clear the state namespace before
starting the topology.
+
### Supported State Backends
#### Redis
@@ -241,49 +265,6 @@ State implementation should be available in the class path
of Storm (by placing
`org.apache.storm:storm-redis:<storm-version>`
-#### HBase
-
-In order to make state scalable, HBaseKeyValueState stores state KV to a row.
This introduces `non-atomic` commit phase and guarantee
-eventual consistency on HBase side. It doesn't matter in point of state's view
because HBaseKeyValueState can still provide not-yet-committed value.
-Even if worker crashes at commit phase, after restart it will read
pending-commit states (stored atomically) from HBase and states will be stored
eventually.
-
-NOTE: HBase state provider uses pre-created table and column family, so users
need to create and provide one to the provider config.
-
-You can simply create table via `create 'state', 'cf'` in `hbase shell` but in
production you may want to give some more properties.
-
-* State provider class name (`topology.state.provider`)
-
-`org.apache.storm.hbase.state.HBaseKeyValueStateProvider`
-
-* Provider config (`topology.state.provider.config`)
-
-```
- {
- "keyClass": "Optional fully qualified class name of the Key type.",
- "valueClass": "Optional fully qualified class name of the Value type.",
- "keySerializerClass": "Optional Key serializer implementation class.",
- "valueSerializerClass": "Optional Value Serializer implementation class.",
- "hbaseConfigKey": "config key to load hbase configuration from storm root
configuration. (similar to storm-hbase)",
- "tableName": "Pre-created table name for state.",
- "columnFamily": "Pre-created column family for state."
- }
- ```
-
-If you want to initialize HBase state provider from codebase, please see below
example:
-
-```
-Config conf = new Config();
- Map<String, Object> hbConf = new HashMap<String, Object>();
- hbConf.put("hbase.rootdir", "file:///tmp/hbase");
- conf.put("hbase.conf", hbConf);
- conf.put("topology.state.provider",
"org.apache.storm.hbase.state.HBaseKeyValueStateProvider");
- conf.put("topology.state.provider.config", "{" +
- " \"hbaseConfigKey\": \"hbase.conf\"," +
- " \"tableName\": \"state\"," +
- " \"columnFamily\": \"cf\"" +
- " }");
-```
-
-* Artifacts to add (`--artifacts`)
-
-`org.apache.storm:storm-hbase:<storm-version>`
\ No newline at end of file
+The HBase state backend was removed in Storm 3.0.0 along with the
`storm-hbase` module (STORM-3988).
+Topologies that used `HBaseKeyValueStateProvider` need to move to another
backend; there is no
+in-place migration, since the stored state is not portable between providers.
\ No newline at end of file
diff --git
a/storm-client/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
b/storm-client/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
index 7633ed59f..75ef038c8 100644
--- a/storm-client/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
+++ b/storm-client/src/jvm/org/apache/storm/state/DefaultStateSerializer.java
@@ -15,6 +15,7 @@ package org.apache.storm.state;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.util.DefaultClassResolver;
import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
import java.util.ArrayList;
@@ -27,8 +28,10 @@ import org.apache.storm.Config;
import org.apache.storm.serialization.KryoTupleDeserializer;
import org.apache.storm.serialization.KryoTupleSerializer;
import org.apache.storm.serialization.SerializationFactory;
+import org.apache.storm.spout.CheckPointState;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ObjectReader;
import org.objenesis.strategy.StdInstantiatorStrategy;
/**
@@ -42,8 +45,18 @@ public class DefaultStateSerializer<T> implements
Serializer<T> {
private final ThreadLocal<Kryo> kryo = new ThreadLocal<Kryo>() {
@Override
protected Kryo initialValue() {
- Kryo obj = new Kryo();
- obj.setRegistrationRequired(false);
+ // Same as new Kryo(), except for the class resolver: the no-arg
constructor is
+ // new Kryo(new DefaultClassResolver(), null).
+ Kryo obj = new Kryo(new StateClassResolver(), null);
+ // Registration bounds the set of classes this serializer will
construct from stored
+ // bytes to the ones the topology declared. It is keyed off the
same config as the tuple
+ // path in DefaultKryoFactory, so a topology that has opted into
permissive Kryo globally
+ // keeps the previous behaviour.
+ // Note this Kryo is independent of the one SerializationFactory
builds for tuples: the
+ // two id spaces must not be merged, since the tuple ids are a
wire format between workers.
+ boolean fallBackOnJavaSerialization = ObjectReader.getBoolean(
+ topoConf == null ? null :
topoConf.get(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION), false);
+ obj.setRegistrationRequired(!fallBackOnJavaSerialization);
if (context != null && topoConf != null) {
KryoTupleSerializer ser = new KryoTupleSerializer(topoConf,
context);
KryoTupleDeserializer deser = new
KryoTupleDeserializer(topoConf, context);
@@ -52,6 +65,7 @@ public class DefaultStateSerializer<T> implements
Serializer<T> {
if (!registrations.isEmpty()) {
SerializationFactory.register(obj, registrations);
}
+ registerInternalClasses(obj);
obj.setInstantiatorStrategy(new DefaultInstantiatorStrategy(new
StdInstantiatorStrategy()));
return obj;
}
@@ -100,6 +114,53 @@ public class DefaultStateSerializer<T> implements
Serializer<T> {
return (T) kryo.get().readClassAndObject(input);
}
+ /**
+ * Registers the types Storm's own state encoding and checkpointing
persist without the component
+ * declaring them.
+ *
+ * <p>Called after the configured registrations so that the ids assigned
to those are unaffected.
+ * {@link Kryo#register(Class)} returns any existing registration, so a
class already declared
+ * through {@link Config#TOPOLOGY_STATE_KRYO_REGISTER} keeps the id
assigned there.
+ *
+ * <p>Deliberately limited to types Storm itself writes without
registering them elsewhere. The
+ * windowing types reachable from a persisted {@code
WindowState.WindowPartition} are registered
+ * by {@code PersistentWindowedBoltExecutor} through {@link
Config#TOPOLOGY_STATE_KRYO_REGISTER}
+ * and must not be added here: that list includes JDK types in {@code
java.base} whose eager
+ * {@code FieldSerializer} construction needs reflective access the module
system denies unless
+ * the worker is started with a matching {@code --add-opens}.
+ */
+ private static void registerInternalClasses(Kryo kryo) {
+ // DefaultStateEncoder wraps every value as Optional<byte[]>.
+ kryo.register(byte[].class);
+ // CheckpointSpout's own state, which it stores without registering.
+ kryo.register(CheckPointState.class);
+ kryo.register(CheckPointState.State.class);
+ }
+
+ /**
+ * A class resolver that always clears its name caches on reset.
+ *
+ * <p>{@link DefaultClassResolver#reset()} returns immediately when
registration is required, on
+ * the assumption that name encoding cannot occur in that mode. That
assumption does not hold
+ * when reading back state: bytes written by an earlier release carry
name-encoded classes, and
+ * Kryo skips the class name in the stream for a name id it believes it
has already seen. Left
+ * uncleared, the second such read on a Kryo instance desynchronises from
the stream and fails
+ * with a buffer underflow rather than reading the value.
+ */
+ private static class StateClassResolver extends DefaultClassResolver {
+ @Override
+ public void reset() {
+ super.reset();
+ if (classToNameId != null) {
+ classToNameId.clear(2048);
+ }
+ if (nameIdToClass != null) {
+ nameIdToClass.clear();
+ }
+ nextNameId = 0;
+ }
+ }
+
private static class TupleSerializer extends
com.esotericsoftware.kryo.Serializer<TupleImpl> {
private final KryoTupleSerializer tupleSerializer;
private final KryoTupleDeserializer tupleDeserializer;
diff --git
a/storm-client/test/jvm/org/apache/storm/state/DefaultStateSerializerTest.java
b/storm-client/test/jvm/org/apache/storm/state/DefaultStateSerializerTest.java
index 20f798c2e..15718b2d4 100644
---
a/storm-client/test/jvm/org/apache/storm/state/DefaultStateSerializerTest.java
+++
b/storm-client/test/jvm/org/apache/storm/state/DefaultStateSerializerTest.java
@@ -18,13 +18,23 @@
package org.apache.storm.state;
+import com.esotericsoftware.kryo.Kryo;
+import com.esotericsoftware.kryo.io.Output;
+import com.esotericsoftware.kryo.util.DefaultInstantiatorStrategy;
import java.util.ArrayList;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import org.apache.storm.Config;
import org.apache.storm.spout.CheckPointState;
import org.junit.jupiter.api.Test;
+import org.objenesis.strategy.StdInstantiatorStrategy;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
/**
* Unit tests for {@link DefaultStateSerializer}
@@ -53,4 +63,111 @@ public class DefaultStateSerializerTest {
assertEquals(cs, s3.deserialize(bytes));
}
+
+ /**
+ * The encoder wraps every value as Optional<byte[]>, so byte[] must
round-trip even though
+ * no caller registers it explicitly.
+ */
+ @Test
+ public void testDefaultStateEncoderRoundTrip() {
+ DefaultStateEncoder<String, byte[]> encoder =
+ new DefaultStateEncoder<>(new DefaultStateSerializer<>(), new
DefaultStateSerializer<>());
+ byte[] value = new byte[]{ 1, 2, 3 };
+
+ assertEquals("k", encoder.decodeKey(encoder.encodeKey("k")));
+ assertArrayEquals(value,
encoder.decodeValue(encoder.encodeValue(value)));
+ // the tombstone is produced by the same static serializer at
class-init time
+ assertNull(encoder.decodeValue(encoder.getTombstoneValue()));
+ }
+
+ @Test
+ public void testDeserializeRejectsUnregisteredClasses() {
+ // a Kryo stream naming a class the topology never registered, as an
earlier release
+ // or an unrelated writer could have left in the store
+ Kryo permissive = new Kryo();
+ permissive.setRegistrationRequired(false);
+ permissive.setInstantiatorStrategy(new DefaultInstantiatorStrategy(new
StdInstantiatorStrategy()));
+ Output out = new Output(4096);
+ permissive.writeClassAndObject(out, new UnregisteredPojo());
+ byte[] unregisteredClassBytes = out.toBytes();
+
+ Serializer<Object> serializer = new DefaultStateSerializer<>();
+ assertThrows(IllegalArgumentException.class, () ->
serializer.deserialize(unregisteredClassBytes));
+ }
+
+ @Test
+ public void testSerializeRejectsUnregisteredClasses() {
+ Serializer<Object> serializer = new DefaultStateSerializer<>();
+ assertThrows(IllegalArgumentException.class, () ->
serializer.serialize(new UnregisteredPojo()));
+ }
+
+ @Test
+ public void testFallBackOnJavaSerializationAllowsUnregisteredClasses() {
+ Map<String, Object> topoConf = new HashMap<>();
+ topoConf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true);
+ Serializer<Object> serializer = new DefaultStateSerializer<>(topoConf,
null);
+ byte[] bytes = serializer.serialize(new UnregisteredPojo());
+ assertEquals(UnregisteredPojo.class,
serializer.deserialize(bytes).getClass());
+ }
+
+ public static class UnregisteredPojo {
+ private long value;
+ }
+
+ /**
+ * Replica of the serializer as it behaved before registration was
required: unregistered classes
+ * are written by name. State persisted by earlier releases is in this
format.
+ */
+ private static Kryo legacyKryo() {
+ Kryo k = new Kryo();
+ k.setRegistrationRequired(false);
+ org.apache.storm.serialization.SerializationFactory.register(
+ k, Collections.singletonList(java.util.Optional.class.getName()));
+ k.setInstantiatorStrategy(new DefaultInstantiatorStrategy(new
StdInstantiatorStrategy()));
+ return k;
+ }
+
+ private static byte[] legacyWrite(Object obj) {
+ Output out = new Output(4096);
+ legacyKryo().writeClassAndObject(out, obj);
+ out.flush();
+ return out.toBytes();
+ }
+
+ /** Exactly what the pre-registration DefaultStateEncoder.encodeValue
produced. */
+ private static byte[] legacyEncodeValue(Object value) {
+ return legacyWrite(java.util.Optional.of(legacyWrite(value)));
+ }
+
+ /**
+ * State written before registration was required is name-encoded. Kryo
skips the class name for
+ * a name id it has already seen, so without clearing that cache between
operations the second
+ * such read desynchronises from the stream.
+ */
+ @Test
+ public void testSequentialReadsOfLegacyEncodedState() {
+ DefaultStateEncoder<String, Object> encoder =
+ new DefaultStateEncoder<>(new DefaultStateSerializer<>(), new
DefaultStateSerializer<>());
+
+ assertEquals("first", encoder.decodeValue(legacyEncodeValue("first")));
+ assertEquals("second",
encoder.decodeValue(legacyEncodeValue("second")));
+ assertEquals("third", encoder.decodeValue(legacyEncodeValue("third")));
+ }
+
+ /**
+ * A rejected payload must not affect later reads. The serializers are
shared, so a failure that
+ * left the class resolver dirty would let one planted value deny service
to the rest.
+ */
+ @Test
+ public void testRejectedPayloadDoesNotBreakLaterReads() {
+ DefaultStateEncoder<String, Object> encoder =
+ new DefaultStateEncoder<>(new DefaultStateSerializer<>(), new
DefaultStateSerializer<>());
+
+ assertThrows(IllegalArgumentException.class,
+ () -> encoder.decodeValue(legacyWrite(new UnregisteredPojo())));
+
+ // a legitimate value, in both the current and the legacy encoding
+ assertEquals("still-works",
encoder.decodeValue(encoder.encodeValue("still-works")));
+ assertEquals("legacy-still-works",
encoder.decodeValue(legacyEncodeValue("legacy-still-works")));
+ }
}
