This is an automated email from the ASF dual-hosted git repository. rzo1 pushed a commit to branch fix/message-id-random-source in repository https://gitbox.apache.org/repos/asf/storm.git
commit 60766b1539e39497df78b3c63b33d525e5d510be Author: Richard Zowalla <[email protected]> AuthorDate: Fri Aug 21 14:30:59 2026 +0200 Generate tuple tree ids from a key stream instead of a recoverable linear congruential generator --- .../jvm/org/apache/storm/executor/Executor.java | 3 +- .../org/apache/storm/utils/KeyStreamRandom.java | 107 +++++++++++++++++++++ .../apache/storm/executor/SpoutExecutorTest.java | 28 ++++++ .../storm/tuple/MessageIdGeneratorBenchmark.java | 53 ++++++++++ .../jvm/org/apache/storm/tuple/MessageIdTest.java | 99 +++++++++++++++++++ 5 files changed, 289 insertions(+), 1 deletion(-) diff --git a/storm-client/src/jvm/org/apache/storm/executor/Executor.java b/storm-client/src/jvm/org/apache/storm/executor/Executor.java index d488a6263..716ec48c4 100644 --- a/storm-client/src/jvm/org/apache/storm/executor/Executor.java +++ b/storm-client/src/jvm/org/apache/storm/executor/Executor.java @@ -84,6 +84,7 @@ import org.apache.storm.tuple.TupleImpl; import org.apache.storm.tuple.Values; import org.apache.storm.utils.ConfigUtils; import org.apache.storm.utils.JCQueue; +import org.apache.storm.utils.KeyStreamRandom; import org.apache.storm.utils.ObjectReader; import org.apache.storm.utils.Time; import org.apache.storm.utils.Utils; @@ -180,7 +181,7 @@ public abstract class Executor implements Callable, JCQueue.Consumer { this.reportErrorDie = new ReportErrorAndDie(reportError, suicideFn); this.sampler = ConfigUtils.mkStatsSampler(topoConf); this.isDebug = ObjectReader.getBoolean(topoConf.get(Config.TOPOLOGY_DEBUG), false); - this.rand = new Random(Utils.secureRandomLong()); + this.rand = new KeyStreamRandom(); this.credentials = credentials; this.hasEventLoggers = StormCommon.hasEventLoggers(topoConf); this.ackingEnabled = StormCommon.hasAckers(topoConf); diff --git a/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java b/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java new file mode 100644 index 000000000..a8c7903f5 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/utils/KeyStreamRandom.java @@ -0,0 +1,107 @@ +/** + * 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.storm.utils; + +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.security.GeneralSecurityException; +import java.security.SecureRandom; +import java.util.Random; +import javax.crypto.Cipher; +import javax.crypto.spec.IvParameterSpec; +import javax.crypto.spec.SecretKeySpec; + +/** + * A {@link Random} that returns slices of an AES counter mode key stream, keyed from {@link SecureRandom} when the instance is + * created and buffered a block at a time. It is meant for values that other parties must not be able to guess, such as the tuple + * tree ids handed out by {@link org.apache.storm.tuple.MessageId#generateId(Random)}: unlike {@link Random}, + * {@link java.util.SplittableRandom} and {@link java.util.concurrent.ThreadLocalRandom}, whose internal state is recoverable from + * a couple of returned values, the values returned here say nothing about the values returned next. + * + * <p>The buffering keeps it cheap enough for the emit path; it is measurably faster than {@link Random}, whose seed update is a + * contended compare and set per value. It is not a drop in replacement for a seeded {@link Random} though, because it cannot be + * reseeded and so cannot produce a repeatable sequence.</p> + */ +public class KeyStreamRandom extends Random { + private static final long serialVersionUID = 1L; + private static final String TRANSFORMATION = "AES/CTR/NoPadding"; + private static final int KEY_BYTES = 16; + private static final int DEFAULT_BUFFER_LONGS = 512; + + private final SecureRandom source = new SecureRandom(); + private final Cipher cipher; + private final byte[] input; + private final byte[] output; + private final long[] buffer; + private int position; + + public KeyStreamRandom() { + this(DEFAULT_BUFFER_LONGS); + } + + KeyStreamRandom(int bufferLongs) { + this.cipher = newCipher(source); + this.input = new byte[bufferLongs * Long.BYTES]; + this.output = new byte[bufferLongs * Long.BYTES]; + this.buffer = new long[bufferLongs]; + // start empty, the first call fills the buffer + this.position = bufferLongs; + } + + private static Cipher newCipher(SecureRandom source) { + byte[] key = new byte[KEY_BYTES]; + byte[] iv = new byte[KEY_BYTES]; + source.nextBytes(key); + source.nextBytes(iv); + try { + Cipher cipher = Cipher.getInstance(TRANSFORMATION); + cipher.init(Cipher.ENCRYPT_MODE, new SecretKeySpec(key, "AES"), new IvParameterSpec(iv)); + return cipher; + } catch (GeneralSecurityException e) { + // no AES here, fall back to taking the values from the SecureRandom itself + return null; + } + } + + @Override + public synchronized long nextLong() { + if (position == buffer.length) { + fill(); + } + return buffer[position++]; + } + + @Override + protected int next(int bits) { + return (int) (nextLong() >>> (Long.SIZE - bits)); + } + + @Override + public void setSeed(long seed) { + // there is nothing to seed, and Random's constructor calls this before this class' fields exist + } + + private void fill() { + if (cipher == null) { + source.nextBytes(output); + } else { + try { + cipher.update(input, 0, input.length, output, 0); + } catch (GeneralSecurityException e) { + throw Utils.wrapInRuntime(e); + } + } + ByteBuffer.wrap(output).order(ByteOrder.nativeOrder()).asLongBuffer().get(buffer); + position = 0; + } +} diff --git a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java index a44597cde..b83009f89 100644 --- a/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java +++ b/storm-client/test/jvm/org/apache/storm/executor/SpoutExecutorTest.java @@ -22,6 +22,7 @@ import org.apache.storm.metrics2.StormMetricRegistry; import org.apache.storm.task.WorkerTopologyContext; import org.apache.storm.tuple.AddressedTuple; import org.apache.storm.tuple.TupleImpl; +import org.apache.storm.utils.KeyStreamRandom; import org.apache.storm.utils.RotatingMap; import org.apache.storm.utils.Utils; import org.junit.jupiter.api.Test; @@ -33,6 +34,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import static org.junit.jupiter.api.Assertions.assertInstanceOf; import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; @@ -84,4 +86,30 @@ public class SpoutExecutorTest { Mockito.verify(rotatingMap,Mockito.times(1)).rotate(); } + + @Test + public void testTupleTreeIdsComeFromAnUnguessableGenerator() { + + RateCounter rateCounter = Mockito.mock(RateCounter.class); + + StormMetricRegistry stormMetricRegistry = Mockito.mock(StormMetricRegistry.class); + Mockito.when(stormMetricRegistry.rateCounter(anyString(),anyString(),anyInt())).thenReturn(rateCounter); + + ComponentCommon componentCommon = Mockito.mock(ComponentCommon.class); + Mockito.when(componentCommon.get_json_conf()).thenReturn(null); + + WorkerTopologyContext workerTopologyContext = Mockito.mock(WorkerTopologyContext.class); + Mockito.when(workerTopologyContext.getComponentId(anyInt())).thenReturn("1"); + Mockito.when(workerTopologyContext.getComponentCommon(anyString())).thenReturn(componentCommon); + + WorkerState workerState = Mockito.mock(WorkerState.class); + Mockito.when(workerState.getWorkerTopologyContext()).thenReturn(workerTopologyContext); + Mockito.when(workerState.getStateStorage()).thenReturn(Mockito.mock(IStateStorage.class)); + Mockito.when(workerState.getTopologyConf()).thenReturn(Utils.readDefaultConfig()); + Mockito.when(workerState.getMetricRegistry()).thenReturn(stormMetricRegistry); + + SpoutExecutor spoutExecutor = new SpoutExecutor(workerState,List.of(1L,5L),new HashMap<>()); + + assertInstanceOf(KeyStreamRandom.class, spoutExecutor.rand); + } } diff --git a/storm-client/test/jvm/org/apache/storm/tuple/MessageIdGeneratorBenchmark.java b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdGeneratorBenchmark.java new file mode 100644 index 000000000..64b1a547c --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdGeneratorBenchmark.java @@ -0,0 +1,53 @@ +/** + * 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.storm.tuple; + +import java.security.SecureRandom; +import java.util.Random; +import org.apache.storm.utils.KeyStreamRandom; + +/** + * Not a test, run it by hand with a main() when the cost of MessageId.generateId is in question. It is a plain timed loop, so + * treat the numbers as an order of magnitude rather than as a measurement. + */ +public final class MessageIdGeneratorBenchmark { + + private static final int WARMUP = 5_000_000; + private static final int ITERATIONS = 50_000_000; + + private MessageIdGeneratorBenchmark() { + } + + public static void main(String[] args) { + for (int round = 0; round < 3; round++) { + System.out.println("round " + round); + time("java.util.Random", new Random(new SecureRandom().nextLong())); + time("KeyStreamRandom", new KeyStreamRandom()); + time("SecureRandom", new SecureRandom()); + } + } + + private static void time(String name, Random rand) { + long sink = 0; + for (int i = 0; i < WARMUP; i++) { + sink ^= MessageId.generateId(rand); + } + long start = System.nanoTime(); + for (int i = 0; i < ITERATIONS; i++) { + sink ^= MessageId.generateId(rand); + } + long elapsed = System.nanoTime() - start; + System.out.printf(" %-22s %6.2f ns/op %,12.0f ops/sec (sink %d)%n", + name, (double) elapsed / ITERATIONS, ITERATIONS / (elapsed / 1e9), sink); + } +} diff --git a/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java new file mode 100644 index 000000000..a8ef0559b --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/tuple/MessageIdTest.java @@ -0,0 +1,99 @@ +/** + * 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.storm.tuple; + +import java.util.HashSet; +import java.util.Random; +import java.util.Set; +import org.apache.storm.utils.KeyStreamRandom; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +public class MessageIdTest { + + private static final long LCG_MULTIPLIER = 0x5DEECE66DL; + private static final long LCG_ADDEND = 0xBL; + private static final long LCG_MASK = (1L << 48) - 1; + + @Test + public void generateIdReturnsDistinctValues() { + Random rand = new KeyStreamRandom(); + Set<Long> ids = new HashSet<>(); + for (int i = 0; i < 100000; i++) { + ids.add(MessageId.generateId(rand)); + } + assertEquals(100000, ids.size()); + } + + @Test + public void generatedIdsDoNotRevealTheFollowingIds() { + Random rand = new KeyStreamRandom(); + for (int i = 0; i < 20; i++) { + long first = MessageId.generateId(rand); + long second = MessageId.generateId(rand); + assertNotEquals(second, predictNextFromLcgState(first)); + } + } + + /** + * Guards the test above: the same prediction does work against a java.util.Random, so a failure of + * generatedIdsDoNotRevealTheFollowingIds means the ids really are predictable rather than the prediction being broken. + */ + @Test + public void lcgPredictionWorksAgainstJavaUtilRandom() { + Random rand = new Random(4242L); + for (int i = 0; i < 20; i++) { + long first = MessageId.generateId(rand); + long second = MessageId.generateId(rand); + assertEquals(second, predictNextFromLcgState(first)); + } + } + + @Test + public void makeRootIdKeepsTheGeneratedIds() { + Random rand = new KeyStreamRandom(); + long id = MessageId.generateId(rand); + long val = MessageId.generateId(rand); + MessageId messageId = MessageId.makeRootId(id, val); + assertEquals(val, messageId.getAnchorsToIds().get(id)); + assertNotNull(messageId.toString()); + } + + /** + * Recovers the 48 bit state of a java.util.Random from a single nextLong output and returns the nextLong it would produce + * afterwards, or 0 if the state could not be recovered. + */ + private static long predictNextFromLcgState(long observed) { + int low = (int) observed; + int high = (int) ((observed - low) >>> 32); + long partialSeed = (high & 0xFFFFFFFFL) << 16; + for (int guess = 0; guess < (1 << 16); guess++) { + long seed = nextSeed(partialSeed | guess); + if ((int) (seed >>> 16) == low) { + seed = nextSeed(seed); + long nextHigh = seed >>> 16; + seed = nextSeed(seed); + long nextLow = seed >>> 16; + return (nextHigh << 32) + (int) nextLow; + } + } + return 0; + } + + private static long nextSeed(long seed) { + return (seed * LCG_MULTIPLIER + LCG_ADDEND) & LCG_MASK; + } +}
