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 7cfc562ea Drop unexpected Pacemaker frames instead of terminating the
daemon; mark Pacemaker as deprecated (#9084)
7cfc562ea is described below
commit 7cfc562ea6cc6e0f5717a56c02e003ff852168ba
Author: Richard Zowalla <[email protected]>
AuthorDate: Tue Sep 15 11:41:15 2026 +0200
Drop unexpected Pacemaker frames instead of terminating the daemon; mark
Pacemaker as deprecated (#9084)
* Drop unexpected Pacemaker frames instead of terminating the daemon
The Pacemaker server decodes a CONTROL_MESSAGE frame into a ControlMessage.
With no SASL handler installed (pacemaker.auth.method NONE), or once the
SASL
handler has forwarded it upstream, the object reached
PacemakerServer.received(),
which cast it to HBMessage. The resulting ClassCastException reached
StormServerHandler.exceptionCaught, which treats anything but an
IOException as
fatal and exits the JVM. A malformed request (undecodable thrift payload, a
control frame without payload or with an unknown code, a request without
data)
took the same path.
The server-side ThriftDecoder now accepts only the control message a
Pacemaker
client sends, SASL_TOKEN_MESSAGE_REQUEST, which starts the DIGEST handshake.
Any other control frame, and control frames whose payload is missing, too
short
or carries an unknown code, is discarded and the connection closed, in the
style of the worker MessageDecoder. SASL_MESSAGE_TOKEN and HBMessage frames
are
decoded as before, so DIGEST and KERBEROS handshakes are unchanged. On the
client side a malformed control frame is reported as an IOException, so that
PacemakerClientHandler reconnects as it did before.
PacemakerServer.received() checks the message type before using it: anything
that is not an HBMessage is logged with its type and remote address and the
connection is closed.
The Pacemaker pipeline uses a new PacemakerServerHandler, a
StormServerHandler
whose exceptionCaught closes the failing connection for any Exception and
keeps serving the other clients. Errors are still handed to
StormServerHandler. StormServerHandler itself, and with it the worker
messaging pipeline, is unchanged.
PacemakerServer gets a package-private close() used by the new tests.
* Mark Pacemaker as deprecated
Pacemaker was built to take worker heartbeat writes off ZooKeeper. Since
STORM-2693 workers heartbeat to their supervisor, which reports them to
Nimbus, so the daemon is only kept for backward compatibility; the metrics
documentation already describes it as deprecated.
Mark the Pacemaker server and client classes, the pacemaker state storage
and the pacemaker.* configuration keys as @Deprecated, and add a notice to
docs/Pacemaker.md and to the storm pacemaker command help.
* Harden Pacemaker SASL token decoding; address review feedback
Bound the declared payload length in SaslMessageToken.read and drop
malformed SASL_MESSAGE_TOKEN frames in the Pacemaker ThriftDecoder
(close the connection on the server, surface an IOException on the
client), mirroring the existing control-frame handling so a malformed
SASL frame can no longer take a large allocation path in the decoder.
Also: add regression tests for the SASL token path, reword the
Pacemaker deprecation note in docs, and document PacemakerServer.close()
as test-only.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
---------
Co-authored-by: Richard Zowalla <[email protected]>
Co-authored-by: Rui Abreu <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
bin/storm.py | 4 +
docs/Pacemaker.md | 7 +
storm-client/src/jvm/org/apache/storm/Config.java | 20 ++
.../storm/cluster/PaceMakerStateStorage.java | 8 +
.../cluster/PaceMakerStateStorageFactory.java | 8 +
.../storm/cluster/StormClusterStateImpl.java | 1 +
.../storm/messaging/netty/SaslMessageToken.java | 5 +-
.../apache/storm/pacemaker/PacemakerClient.java | 8 +
.../storm/pacemaker/PacemakerClientHandler.java | 8 +
.../storm/pacemaker/PacemakerClientPool.java | 8 +
.../pacemaker/PacemakerConnectionException.java | 8 +
.../storm/pacemaker/codec/ThriftDecoder.java | 80 ++++++-
.../storm/pacemaker/codec/ThriftEncoder.java | 8 +
.../pacemaker/codec/ThriftNettyClientCodec.java | 8 +
.../storm/PaceMakerStateStorageFactoryTest.java | 1 +
.../storm/pacemaker/codec/ThriftDecoderTest.java | 194 +++++++++++++++
.../main/java/org/apache/storm/DaemonConfig.java | 8 +
.../storm/pacemaker/IServerMessageHandler.java | 8 +
.../java/org/apache/storm/pacemaker/Pacemaker.java | 8 +
.../apache/storm/pacemaker/PacemakerServer.java | 24 ++
.../pacemaker/codec/PacemakerServerHandler.java | 50 ++++
.../pacemaker/codec/ThriftNettyServerCodec.java | 13 +-
.../test/java/org/apache/storm/PacemakerTest.java | 1 +
.../storm/pacemaker/PacemakerServerTest.java | 261 +++++++++++++++++++++
24 files changed, 743 insertions(+), 6 deletions(-)
diff --git a/bin/storm.py b/bin/storm.py
index 55c211df8..f8c7ab3f1 100755
--- a/bin/storm.py
+++ b/bin/storm.py
@@ -925,6 +925,10 @@ def initialize_pacemaker_subcommand(subparsers):
Launches the Pacemaker daemon. This command should be run under
supervision with a tool like daemontools or monit.
+ Deprecated: Pacemaker is only kept for backward compatibility and will be
+ removed in a future release. By default workers heartbeat to their
+ supervisor, which reports them to Nimbus; no Pacemaker is needed.
+
See Setting up a Storm cluster for more information.
(https://storm.apache.org/documentation/Setting-up-a-Storm-cluster)
"""
diff --git a/docs/Pacemaker.md b/docs/Pacemaker.md
index 75fa74855..4e496a504 100644
--- a/docs/Pacemaker.md
+++ b/docs/Pacemaker.md
@@ -4,6 +4,13 @@ layout: documentation
documentation: true
---
+> **Deprecated:** Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+> Since Storm 2.0
([STORM-2693](https://issues.apache.org/jira/browse/STORM-2693)) workers no
longer heartbeat directly into
+> ZooKeeper: each worker writes its heartbeats to local disk and its
supervisor reports them to Nimbus over Thrift (see
+> [Daemon Fault Tolerance](Daemon-Fault-Tolerance.html)). This removes the
ZooKeeper write load Pacemaker was built to avoid.
+> New clusters should keep the default `storm.cluster.state.store:
"org.apache.storm.cluster.ZKStateStorageFactory"` and
+> not run Pacemaker; existing deployments should plan to migrate away from
Pacemaker back to the default
+> ZooKeeper-based state store.
### Introduction
Pacemaker is a storm daemon designed to process heartbeats from workers. As
Storm is scaled up, ZooKeeper begins to become a bottleneck due to high volumes
of writes from workers doing heartbeats. Lots of writes to disk and too much
traffic across the network is generated as ZooKeeper tries to maintain
consistency.
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 53f8eefdf..519b8d943 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1119,36 +1119,56 @@ public class Config extends HashMap<String, Object> {
public static final String TOPOLOGY_MAX_REPLICATION_WAIT_TIME_SEC =
"topology.max.replication.wait.time.sec";
/**
* The list of servers that Pacemaker is running on.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsStringList
+ @Deprecated
public static final String PACEMAKER_SERVERS = "pacemaker.servers";
/**
* The port Pacemaker should run on. Clients should connect to this port
to submit or read heartbeats.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsNumber
@IsPositiveNumber
+ @Deprecated
public static final String PACEMAKER_PORT = "pacemaker.port";
/**
* The maximum number of threads that should be used by the Pacemaker
client.
* When Pacemaker gets loaded it will spawn new threads, up to
* this many total, to handle the load.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsNumber
@IsPositiveNumber
+ @Deprecated
public static final String PACEMAKER_CLIENT_MAX_THREADS =
"pacemaker.client.max.threads";
/**
* This should be one of "DIGEST", "KERBEROS", or "NONE" Determines the
mode of authentication the pacemaker server and client use. The
* client must either match the server, or be NONE. In the case of NONE,
no authentication is performed for the client, and if the
* server is running with DIGEST or KERBEROS, the client can only write to
the server (no reads). This is intended to provide a
* primitive form of access-control.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@CustomValidator(validatorClass =
ConfigValidation.PacemakerAuthTypeValidator.class)
+ @Deprecated
public static final String PACEMAKER_AUTH_METHOD = "pacemaker.auth.method";
/**
* Pacemaker Thrift Max Message Size (bytes).
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsInteger
@IsPositiveNumber
+ @Deprecated
public static final String PACEMAKER_THRIFT_MESSAGE_SIZE_MAX =
"pacemaker.thrift.message.size.max";
/**
* Max no.of seconds group mapping service will cache user groups
diff --git
a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java
b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java
index e746e5ec6..effaeea11 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorage.java
@@ -37,6 +37,14 @@ import org.apache.storm.utils.WrappedHBExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * State storage that keeps worker heartbeats in Pacemaker and everything else
in ZooKeeper.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PaceMakerStateStorage implements IStateStorage {
private static final int maxRetries = 10;
diff --git
a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
index 596aefab6..6b8e3bf32 100644
---
a/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
+++
b/storm-client/src/jvm/org/apache/storm/cluster/PaceMakerStateStorageFactory.java
@@ -22,6 +22,14 @@ import java.util.Map;
import org.apache.storm.pacemaker.PacemakerClientPool;
import org.apache.storm.utils.Utils;
+/**
+ * Factory for {@link PaceMakerStateStorage}.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PaceMakerStateStorageFactory implements StateStorageFactory {
@Override
public IStateStorage mkStore(Map<String, Object> config, Map<String,
Object> authConf, ClusterStateContext context) {
diff --git
a/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
b/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
index 3044b37ca..8be795c2b 100644
--- a/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
+++ b/storm-client/src/jvm/org/apache/storm/cluster/StormClusterStateImpl.java
@@ -230,6 +230,7 @@ public class StormClusterStateImpl implements
IStormClusterState {
}
@Override
+ @SuppressWarnings("deprecation")
public boolean isPacemakerStateStore() {
if (stateStorage == null) {
return false;
diff --git
a/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslMessageToken.java
b/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslMessageToken.java
index 368ffba02..7e8ecf20e 100644
---
a/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslMessageToken.java
+++
b/storm-client/src/jvm/org/apache/storm/messaging/netty/SaslMessageToken.java
@@ -47,9 +47,12 @@ public class SaslMessageToken implements INettySerializable {
public static SaslMessageToken read(byte[] serial) {
ByteBuf smBuffer = Unpooled.wrappedBuffer(serial);
try {
+ if (smBuffer.readableBytes() < 6) {
+ return null;
+ }
short identifier = smBuffer.readShort();
int payloadLen = smBuffer.readInt();
- if (identifier != IDENTIFIER) {
+ if (identifier != IDENTIFIER || payloadLen < 0 || payloadLen >
smBuffer.readableBytes()) {
return null;
}
byte[] token = new byte[payloadLen];
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
index 4d8e6b528..b4a4b66b3 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -39,6 +39,14 @@ import
org.apache.storm.utils.StormBoundedExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Netty client that sends heartbeat requests to a single Pacemaker server.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PacemakerClient implements ISaslClient {
private static final Logger LOG =
LoggerFactory.getLogger(PacemakerClient.class);
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java
index 31fabc1a1..c3e183cd8 100644
---
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java
+++
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientHandler.java
@@ -21,6 +21,14 @@ import
org.apache.storm.shade.io.netty.channel.ChannelInboundHandlerAdapter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Inbound handler of the Pacemaker client pipeline.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PacemakerClientHandler extends ChannelInboundHandlerAdapter {
private static final Logger LOG =
LoggerFactory.getLogger(PacemakerClientHandler.class);
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
index e17c77b4c..43c6c1f7c 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClientPool.java
@@ -23,6 +23,14 @@ import org.apache.storm.generated.HBMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Pool of clients for the configured Pacemaker servers.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PacemakerClientPool {
private static final Logger LOG =
LoggerFactory.getLogger(PacemakerClientPool.class);
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerConnectionException.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerConnectionException.java
index aedcbcf7b..3390cc5a8 100644
---
a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerConnectionException.java
+++
b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerConnectionException.java
@@ -12,6 +12,14 @@
package org.apache.storm.pacemaker;
+/**
+ * Thrown when no connection to a Pacemaker server is available.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class PacemakerConnectionException extends Exception {
public PacemakerConnectionException(String err) {
super(err);
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
index ce104ed1e..b66420e5c 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
@@ -22,9 +22,20 @@ import org.apache.storm.shade.io.netty.buffer.ByteBuf;
import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext;
import org.apache.storm.shade.io.netty.handler.codec.ByteToMessageDecoder;
import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+/**
+ * Decodes length-prefixed thrift {@link HBMessage} frames of the Pacemaker
protocol.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class ThriftDecoder extends ByteToMessageDecoder {
+ private static final Logger LOG =
LoggerFactory.getLogger(ThriftDecoder.class);
private static final int INTEGER_SIZE = 4;
/**
@@ -32,11 +43,28 @@ public class ThriftDecoder extends ByteToMessageDecoder {
*/
private final int maxLength;
+ /**
+ * Whether this decoder sits in a Pacemaker server pipeline. A server only
accepts the control message a client
+ * sends to start the SASL handshake; any other control frame is dropped
and the connection closed.
+ */
+ private final boolean serverSide;
+
/**
* Instantiate a ThriftDecoder that accepts serialized messages of at most
maxLength bytes.
*/
public ThriftDecoder(final int maxLengthBytes) {
+ this(maxLengthBytes, false);
+ }
+
+ /**
+ * Instantiate a ThriftDecoder that accepts serialized messages of at most
maxLength bytes.
+ *
+ * @param maxLengthBytes the maximum length of a serialized thrift message
+ * @param serverSide true if the decoder is used by a Pacemaker server,
which restricts the control messages it accepts
+ */
+ public ThriftDecoder(final int maxLengthBytes, final boolean serverSide) {
maxLength = maxLengthBytes;
+ this.serverSide = serverSide;
}
@Override
@@ -68,13 +96,61 @@ public class ThriftDecoder extends ByteToMessageDecoder {
HBMessage m = (HBMessage) Utils.thriftDeserialize(HBMessage.class,
serialized);
if (m.get_type() == HBServerMessageType.CONTROL_MESSAGE) {
- ControlMessage cm =
ControlMessage.read(m.get_data().get_message_blob());
+ ControlMessage cm = readControlMessage(m);
+ if (cm == null) {
+ if (!serverSide) {
+ // Let the client handler see the failure so that it
reconnects.
+ throw new IOException("Received a malformed control
message");
+ }
+ dropAndClose(channelHandlerContext, buf, "a malformed control
frame");
+ return;
+ }
+ if (serverSide && cm != ControlMessage.SASL_TOKEN_MESSAGE_REQUEST)
{
+ dropAndClose(channelHandlerContext, buf, "an unexpected
control frame " + cm);
+ return;
+ }
out.add(cm);
} else if (m.get_type() == HBServerMessageType.SASL_MESSAGE_TOKEN) {
- SaslMessageToken sm =
SaslMessageToken.read(m.get_data().get_message_blob());
+ SaslMessageToken sm = readSaslMessageToken(m);
+ if (sm == null) {
+ if (!serverSide) {
+ // Let the client handler see the failure so that it
reconnects.
+ throw new IOException("Received a malformed SASL token
message");
+ }
+ dropAndClose(channelHandlerContext, buf, "a malformed SASL
token frame");
+ return;
+ }
out.add(sm);
} else {
out.add(m);
}
}
+
+ private static ControlMessage readControlMessage(HBMessage m) {
+ if (m.get_data() == null || !m.get_data().is_set_message_blob()) {
+ return null;
+ }
+ byte[] blob = m.get_data().get_message_blob();
+ if (blob == null || blob.length < 2) {
+ return null;
+ }
+ return ControlMessage.read(blob);
+ }
+
+ private static SaslMessageToken readSaslMessageToken(HBMessage m) {
+ if (m.get_data() == null || !m.get_data().is_set_message_blob()) {
+ return null;
+ }
+ byte[] blob = m.get_data().get_message_blob();
+ if (blob == null) {
+ return null;
+ }
+ return SaslMessageToken.read(blob);
+ }
+
+ private static void dropAndClose(ChannelHandlerContext ctx, ByteBuf buf,
String what) {
+ LOG.warn("Channel {} sent {}; closing the connection", ctx.channel(),
what);
+ buf.skipBytes(buf.readableBytes());
+ ctx.close();
+ }
}
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftEncoder.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftEncoder.java
index 659f05166..d029888f1 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftEncoder.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftEncoder.java
@@ -28,6 +28,14 @@ import org.apache.storm.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Encodes messages as length-prefixed thrift {@link HBMessage} frames of the
Pacemaker protocol.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class ThriftEncoder extends MessageToMessageEncoder<Object> {
private static final Logger LOG = LoggerFactory
diff --git
a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
index b20883437..55f94d5da 100644
---
a/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
+++
b/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
@@ -25,6 +25,14 @@ import
org.apache.storm.shade.io.netty.channel.ChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Builds the Pacemaker client pipeline.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class ThriftNettyClientCodec extends ChannelInitializer<Channel> {
public static final String SASL_HANDLER = "sasl-handler";
diff --git
a/storm-client/test/jvm/org/apache/storm/PaceMakerStateStorageFactoryTest.java
b/storm-client/test/jvm/org/apache/storm/PaceMakerStateStorageFactoryTest.java
index 8b482e823..65590d41d 100644
---
a/storm-client/test/jvm/org/apache/storm/PaceMakerStateStorageFactoryTest.java
+++
b/storm-client/test/jvm/org/apache/storm/PaceMakerStateStorageFactoryTest.java
@@ -48,6 +48,7 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;
@ExtendWith(MockitoExtension.class)
+@SuppressWarnings("deprecation")
public class PaceMakerStateStorageFactoryTest {
@Captor
diff --git
a/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java
b/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java
new file mode 100644
index 000000000..3461037a4
--- /dev/null
+++
b/storm-client/test/jvm/org/apache/storm/pacemaker/codec/ThriftDecoderTest.java
@@ -0,0 +1,194 @@
+/**
+ * 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 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.pacemaker.codec;
+
+import org.apache.storm.generated.HBMessage;
+import org.apache.storm.generated.HBMessageData;
+import org.apache.storm.generated.HBServerMessageType;
+import org.apache.storm.messaging.netty.ControlMessage;
+import org.apache.storm.messaging.netty.SaslMessageToken;
+import org.apache.storm.shade.io.netty.buffer.ByteBuf;
+import org.apache.storm.shade.io.netty.buffer.Unpooled;
+import org.apache.storm.shade.io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.storm.shade.io.netty.handler.codec.DecoderException;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+@SuppressWarnings("deprecation")
+public class ThriftDecoderTest {
+
+ private static final int MAX_LENGTH = 1024 * 1024;
+
+ static ByteBuf frame(HBMessage message) {
+ byte[] serialized = Utils.thriftSerialize(message);
+ ByteBuf buf = Unpooled.buffer();
+ buf.writeInt(serialized.length);
+ buf.writeBytes(serialized);
+ return buf;
+ }
+
+ static ByteBuf controlFrame(ControlMessage controlMessage) {
+ ByteBuf blob = Unpooled.buffer();
+ controlMessage.write(blob);
+ byte[] bytes = new byte[blob.readableBytes()];
+ blob.readBytes(bytes);
+ return frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE,
HBMessageData.message_blob(bytes)));
+ }
+
+ private static EmbeddedChannel serverChannel() {
+ return new EmbeddedChannel(new ThriftDecoder(MAX_LENGTH, true));
+ }
+
+ static ByteBuf saslTokenFrame(short identifier, int declaredPayloadLen,
byte[] payload) {
+ ByteBuf blob = Unpooled.buffer();
+ blob.writeShort(identifier);
+ blob.writeInt(declaredPayloadLen);
+ if (payload != null) {
+ blob.writeBytes(payload);
+ }
+ byte[] bytes = new byte[blob.readableBytes()];
+ blob.readBytes(bytes);
+ return frame(new HBMessage(HBServerMessageType.SASL_MESSAGE_TOKEN,
HBMessageData.message_blob(bytes)));
+ }
+
+ @Test
+ public void serverDropsSaslTokenWithOversizedPayloadLength() {
+ EmbeddedChannel channel = serverChannel();
+
+ // A tiny frame that claims a ~2GB payload must not be allocated; it
is dropped and the connection closed.
+ channel.writeInbound(saslTokenFrame(SaslMessageToken.IDENTIFIER,
Integer.MAX_VALUE, null));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void clientReportsSaslTokenWithOversizedPayloadLength() {
+ EmbeddedChannel channel = new EmbeddedChannel(new
ThriftDecoder(MAX_LENGTH));
+
+ assertThrows(DecoderException.class, () -> channel.writeInbound(
+ saslTokenFrame(SaslMessageToken.IDENTIFIER, Integer.MAX_VALUE,
null)));
+ assertNull(channel.readInbound());
+ }
+
+ @Test
+ public void serverDropsUnexpectedControlFrames() {
+ for (ControlMessage controlMessage : ControlMessage.values()) {
+ if (controlMessage == ControlMessage.SASL_TOKEN_MESSAGE_REQUEST) {
+ continue;
+ }
+ EmbeddedChannel channel = serverChannel();
+
+ channel.writeInbound(controlFrame(controlMessage));
+
+ assertNull(channel.readInbound(), controlMessage.name());
+ assertFalse(channel.isActive(), controlMessage.name());
+ }
+ }
+
+ @Test
+ public void serverAcceptsSaslTokenMessageRequest() {
+ EmbeddedChannel channel = serverChannel();
+
+
channel.writeInbound(controlFrame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST));
+
+ assertSame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST,
channel.readInbound());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void serverAcceptsSaslMessageToken() {
+ EmbeddedChannel channel = serverChannel();
+ byte[] token = { 1, 2, 3 };
+ ByteBuf blob = Unpooled.buffer();
+ new SaslMessageToken(token).write(blob);
+ byte[] bytes = new byte[blob.readableBytes()];
+ blob.readBytes(bytes);
+
+ channel.writeInbound(frame(new
HBMessage(HBServerMessageType.SASL_MESSAGE_TOKEN,
HBMessageData.message_blob(bytes))));
+
+ SaslMessageToken decoded = channel.readInbound();
+ assertArrayEquals(token, decoded.getSaslToken());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void serverPassesHeartbeatMessages() {
+ EmbeddedChannel channel = serverChannel();
+ HBMessage message = new HBMessage(HBServerMessageType.CREATE_PATH,
HBMessageData.path("/path"));
+
+ channel.writeInbound(frame(message));
+
+ assertEquals(message, channel.readInbound());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void serverDropsControlFrameWithUnknownCode() {
+ EmbeddedChannel channel = serverChannel();
+
+ channel.writeInbound(frame(new
HBMessage(HBServerMessageType.CONTROL_MESSAGE,
+
HBMessageData.message_blob(new byte[]{ 0, 1 }))));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void serverDropsControlFrameWithoutPayload() {
+ EmbeddedChannel channel = serverChannel();
+
+ channel.writeInbound(frame(new
HBMessage(HBServerMessageType.CONTROL_MESSAGE, null)));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void serverDropsControlFrameWithShortPayload() {
+ EmbeddedChannel channel = serverChannel();
+
+ channel.writeInbound(frame(new
HBMessage(HBServerMessageType.CONTROL_MESSAGE,
+
HBMessageData.message_blob(new byte[]{ 1 }))));
+
+ assertNull(channel.readInbound());
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void clientDecodesSaslCompleteRequest() {
+ EmbeddedChannel channel = new EmbeddedChannel(new
ThriftDecoder(MAX_LENGTH));
+
+
channel.writeInbound(controlFrame(ControlMessage.SASL_COMPLETE_REQUEST));
+
+ assertSame(ControlMessage.SASL_COMPLETE_REQUEST,
channel.readInbound());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void clientReportsMalformedControlFrame() {
+ EmbeddedChannel channel = new EmbeddedChannel(new
ThriftDecoder(MAX_LENGTH));
+
+ assertThrows(DecoderException.class, () -> channel.writeInbound(
+ frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE, null))));
+ assertNull(channel.readInbound());
+ }
+}
diff --git a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
index 4805f5195..e535baf1b 100644
--- a/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
+++ b/storm-server/src/main/java/org/apache/storm/DaemonConfig.java
@@ -682,15 +682,23 @@ public class DaemonConfig implements Validated {
/**
* The maximum number of threads that should be used by the Pacemaker.
When Pacemaker gets loaded it will spawn new threads, up to this
* many total, to handle the load.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsNumber
@IsPositiveNumber
+ @Deprecated
public static final String PACEMAKER_MAX_THREADS = "pacemaker.max.threads";
/**
* This parameter is used by the storm-deploy project to configure the jvm
options for the pacemaker daemon.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path (workers heartbeat to their
supervisor, which reports them to Nimbus) instead.
*/
@IsStringOrStringList
+ @Deprecated
public static final String PACEMAKER_CHILDOPTS = "pacemaker.childopts";
diff --git
a/storm-server/src/main/java/org/apache/storm/pacemaker/IServerMessageHandler.java
b/storm-server/src/main/java/org/apache/storm/pacemaker/IServerMessageHandler.java
index 73451e9cf..dac6d4442 100644
---
a/storm-server/src/main/java/org/apache/storm/pacemaker/IServerMessageHandler.java
+++
b/storm-server/src/main/java/org/apache/storm/pacemaker/IServerMessageHandler.java
@@ -14,6 +14,14 @@ package org.apache.storm.pacemaker;
import org.apache.storm.generated.HBMessage;
+/**
+ * Handles heartbeat requests received by a Pacemaker server.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public interface IServerMessageHandler {
HBMessage handleMessage(HBMessage m, boolean authenticated);
diff --git
a/storm-server/src/main/java/org/apache/storm/pacemaker/Pacemaker.java
b/storm-server/src/main/java/org/apache/storm/pacemaker/Pacemaker.java
index 51441512c..a7809c407 100644
--- a/storm-server/src/main/java/org/apache/storm/pacemaker/Pacemaker.java
+++ b/storm-server/src/main/java/org/apache/storm/pacemaker/Pacemaker.java
@@ -33,6 +33,14 @@ import org.apache.storm.utils.VersionInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * The Pacemaker daemon, an in-memory store for worker heartbeats.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class Pacemaker implements IServerMessageHandler {
private static final Logger LOG = LoggerFactory.getLogger(Pacemaker.class);
diff --git
a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
index 19e6cbe69..123f77aac 100644
--- a/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
+++ b/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
@@ -38,6 +38,14 @@ import
org.apache.storm.shade.io.netty.util.concurrent.GlobalEventExecutor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Netty server of the Pacemaker daemon.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
class PacemakerServer implements ISaslServer {
private static final int FIVE_MB_IN_BYTES = 5 * 1024 * 1024;
@@ -133,6 +141,12 @@ class PacemakerServer implements ISaslServer {
@Override
public void received(Object mesg, String remote, Channel channel) throws
InterruptedException {
+ if (!(mesg instanceof HBMessage)) {
+ LOG.warn("Dropping unexpected message of type {} from {}; closing
the connection",
+ mesg == null ? null : mesg.getClass().getName(), remote);
+ channel.close();
+ return;
+ }
cleanPipeline(channel);
boolean authenticated = (authMethod ==
ThriftNettyServerCodec.AuthMethod.NONE) ||
authenticatedChannels.contains(channel);
@@ -148,6 +162,16 @@ class PacemakerServer implements ISaslServer {
}
}
+ /**
+ * Close all channels and stop the event loops of this server. The
Pacemaker daemon itself runs until the JVM exits,
+ * so this exists for tests to shut a server down deterministically.
+ */
+ void close() {
+ allChannels.close().awaitUninterruptibly();
+ bossEventLoopGroup.shutdownGracefully().awaitUninterruptibly();
+ workerEventLoopGroup.shutdownGracefully().awaitUninterruptibly();
+ }
+
@Override
public String name() {
return topologyName;
diff --git
a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java
b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java
new file mode 100644
index 000000000..7bd8b1a82
--- /dev/null
+++
b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/PacemakerServerHandler.java
@@ -0,0 +1,50 @@
+/**
+ * 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.pacemaker.codec;
+
+import org.apache.storm.messaging.netty.IServer;
+import org.apache.storm.messaging.netty.StormServerHandler;
+import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Pacemaker server handler. A failure while handling a request only affects
the connection it arrived on: the
+ * connection is closed and the Pacemaker server keeps serving its other
clients. Errors are still handled by
+ * {@link StormServerHandler}.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
+public class PacemakerServerHandler extends StormServerHandler {
+ private static final Logger LOG =
LoggerFactory.getLogger(PacemakerServerHandler.class);
+
+ public PacemakerServerHandler(IServer server) {
+ super(server);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (!(cause instanceof Exception)) {
+ super.exceptionCaught(ctx, cause);
+ return;
+ }
+ try {
+ LOG.warn("Closing connection {} after failing to handle its
request", ctx.channel(), cause);
+ } finally {
+ ctx.close();
+ }
+ }
+}
diff --git
a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
index d15f369d9..2b93372f6 100644
---
a/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
+++
b/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
@@ -21,7 +21,6 @@ import org.apache.storm.messaging.netty.ISaslServer;
import org.apache.storm.messaging.netty.IServer;
import org.apache.storm.messaging.netty.KerberosSaslServerHandler;
import org.apache.storm.messaging.netty.SaslStormServerHandler;
-import org.apache.storm.messaging.netty.StormServerHandler;
import org.apache.storm.security.auth.ClientAuthUtils;
import org.apache.storm.shade.io.netty.channel.Channel;
import org.apache.storm.shade.io.netty.channel.ChannelInitializer;
@@ -29,6 +28,14 @@ import
org.apache.storm.shade.io.netty.channel.ChannelPipeline;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+/**
+ * Builds the Pacemaker server pipeline.
+ *
+ * @deprecated Pacemaker is deprecated and only kept for backward
compatibility; it will be removed in a future release.
+ * Use the default heartbeat path instead: workers heartbeat to their
supervisor, which reports them to Nimbus over
+ * Thrift, with the default ZooKeeper-based cluster state store ({@code
org.apache.storm.cluster.ZKStateStorageFactory}).
+ */
+@Deprecated
public class ThriftNettyServerCodec extends ChannelInitializer<Channel> {
public static final String SASL_HANDLER = "sasl-handler";
@@ -54,7 +61,7 @@ public class ThriftNettyServerCodec extends
ChannelInitializer<Channel> {
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("encoder", new ThriftEncoder());
- pipeline.addLast("decoder", new
ThriftDecoder(thriftMessageMaxSizeBytes));
+ pipeline.addLast("decoder", new
ThriftDecoder(thriftMessageMaxSizeBytes, true));
if (authMethod == AuthMethod.DIGEST) {
try {
LOG.debug("Adding SaslStormServerHandler to pacemaker server
pipeline.");
@@ -78,7 +85,7 @@ public class ThriftNettyServerCodec extends
ChannelInitializer<Channel> {
LOG.debug("Not authenticating any clients. AuthMethod is NONE");
}
- pipeline.addLast("handler", new StormServerHandler(server));
+ pipeline.addLast("handler", new PacemakerServerHandler(server));
}
public enum AuthMethod {
diff --git a/storm-server/src/test/java/org/apache/storm/PacemakerTest.java
b/storm-server/src/test/java/org/apache/storm/PacemakerTest.java
index e8ea2a441..9fd5505f8 100644
--- a/storm-server/src/test/java/org/apache/storm/PacemakerTest.java
+++ b/storm-server/src/test/java/org/apache/storm/PacemakerTest.java
@@ -32,6 +32,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
+@SuppressWarnings("deprecation")
public class PacemakerTest {
private HBMessage hbMessage;
diff --git
a/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java
b/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java
new file mode 100644
index 000000000..1718d64ad
--- /dev/null
+++
b/storm-server/src/test/java/org/apache/storm/pacemaker/PacemakerServerTest.java
@@ -0,0 +1,261 @@
+/**
+ * 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.pacemaker;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.storm.Config;
+import org.apache.storm.DaemonConfig;
+import org.apache.storm.generated.HBMessage;
+import org.apache.storm.generated.HBMessageData;
+import org.apache.storm.generated.HBServerMessageType;
+import org.apache.storm.messaging.netty.ControlMessage;
+import org.apache.storm.messaging.netty.ISaslClient;
+import org.apache.storm.messaging.netty.ISaslServer;
+import org.apache.storm.messaging.netty.SaslMessageToken;
+import org.apache.storm.messaging.netty.SaslStormClientHandler;
+import org.apache.storm.metric.StormMetricsRegistry;
+import org.apache.storm.pacemaker.codec.ThriftDecoder;
+import org.apache.storm.pacemaker.codec.ThriftEncoder;
+import org.apache.storm.pacemaker.codec.ThriftNettyServerCodec;
+import org.apache.storm.pacemaker.codec.ThriftNettyServerCodec.AuthMethod;
+import org.apache.storm.shade.io.netty.buffer.ByteBuf;
+import org.apache.storm.shade.io.netty.buffer.Unpooled;
+import org.apache.storm.shade.io.netty.channel.Channel;
+import org.apache.storm.shade.io.netty.channel.embedded.EmbeddedChannel;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.atLeastOnce;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@SuppressWarnings("deprecation")
+public class PacemakerServerTest {
+
+ private static final int MAX_LENGTH = 1024 * 1024;
+
+ private static PacemakerServer server;
+
+ private static Map<String, Object> config() {
+ Map<String, Object> conf = new HashMap<>();
+ conf.put(Config.PACEMAKER_PORT, 0);
+ conf.put(Config.PACEMAKER_AUTH_METHOD, "NONE");
+ conf.put(DaemonConfig.PACEMAKER_MAX_THREADS, 1);
+ conf.put(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX, MAX_LENGTH);
+ return conf;
+ }
+
+ private static ByteBuf frame(byte[] serialized) {
+ ByteBuf buf = Unpooled.buffer();
+ buf.writeInt(serialized.length);
+ buf.writeBytes(serialized);
+ return buf;
+ }
+
+ private static ByteBuf frame(HBMessage message) {
+ return frame(Utils.thriftSerialize(message));
+ }
+
+ private static ByteBuf controlFrame(ControlMessage controlMessage) {
+ ByteBuf buf = Unpooled.buffer();
+ controlMessage.write(buf);
+ byte[] blob = new byte[buf.readableBytes()];
+ buf.readBytes(blob);
+ return frame(new HBMessage(HBServerMessageType.CONTROL_MESSAGE,
HBMessageData.message_blob(blob)));
+ }
+
+ private static ByteBuf saslTokenFrame(short identifier, int
declaredPayloadLen, byte[] payload) {
+ ByteBuf buf = Unpooled.buffer();
+ buf.writeShort(identifier);
+ buf.writeInt(declaredPayloadLen);
+ if (payload != null) {
+ buf.writeBytes(payload);
+ }
+ byte[] blob = new byte[buf.readableBytes()];
+ buf.readBytes(blob);
+ return frame(new HBMessage(HBServerMessageType.SASL_MESSAGE_TOKEN,
HBMessageData.message_blob(blob)));
+ }
+
+ private static HBMessage readResponse(EmbeddedChannel serverChannel) {
+ EmbeddedChannel decoder = new EmbeddedChannel(new
ThriftDecoder(MAX_LENGTH));
+ Object out;
+ while ((out = serverChannel.readOutbound()) != null) {
+ decoder.writeInbound(out);
+ }
+ return decoder.readInbound();
+ }
+
+ private EmbeddedChannel pipeline() {
+ return new EmbeddedChannel(new ThriftNettyServerCodec(server,
config(), AuthMethod.NONE, MAX_LENGTH));
+ }
+
+ @BeforeAll
+ public static void setUp() {
+ server = new PacemakerServer(new Pacemaker(new ConcurrentHashMap<>(),
new StormMetricsRegistry()), config());
+ }
+
+ @AfterAll
+ public static void tearDown() {
+ server.close();
+ }
+
+ @Test
+ public void heartbeatRequestIsAnswered() {
+ EmbeddedChannel channel = pipeline();
+ HBMessage request = new HBMessage(HBServerMessageType.CREATE_PATH,
HBMessageData.path("/path"));
+ request.set_message_id(7);
+
+ channel.writeInbound(frame(request));
+
+ HBMessage response = readResponse(channel);
+ assertEquals(HBServerMessageType.CREATE_PATH_RESPONSE,
response.get_type());
+ assertEquals(7, response.get_message_id());
+ assertTrue(channel.isActive());
+ }
+
+ @Test
+ public void controlFrameClosesOnlyThatConnection() {
+ EmbeddedChannel other = pipeline();
+ EmbeddedChannel channel = pipeline();
+
+ channel.writeInbound(controlFrame(ControlMessage.CLOSE_MESSAGE));
+
+ assertFalse(channel.isActive());
+ assertNull(channel.readOutbound());
+
+ other.writeInbound(frame(new
HBMessage(HBServerMessageType.CREATE_PATH, HBMessageData.path("/path"))));
+ assertEquals(HBServerMessageType.CREATE_PATH_RESPONSE,
readResponse(other).get_type());
+ assertTrue(other.isActive());
+ }
+
+ @Test
+ public void oversizedSaslTokenClosesOnlyThatConnection() {
+ EmbeddedChannel other = pipeline();
+ EmbeddedChannel channel = pipeline();
+
+ // A SASL token frame that declares a ~2GB payload previously OOM'd
the decoder and terminated the daemon.
+ channel.writeInbound(saslTokenFrame(SaslMessageToken.IDENTIFIER,
Integer.MAX_VALUE, null));
+
+ assertFalse(channel.isActive());
+ assertNull(channel.readOutbound());
+
+ other.writeInbound(frame(new
HBMessage(HBServerMessageType.CREATE_PATH, HBMessageData.path("/path"))));
+ assertEquals(HBServerMessageType.CREATE_PATH_RESPONSE,
readResponse(other).get_type());
+ assertTrue(other.isActive());
+ }
+
+ @Test
+ public void saslFrameWithoutAuthenticationConfiguredClosesConnection() {
+ // Passes the decoder, but no SASL handler is installed when the auth
method is NONE.
+ EmbeddedChannel channel = pipeline();
+
+
channel.writeInbound(controlFrame(ControlMessage.SASL_TOKEN_MESSAGE_REQUEST));
+
+ assertFalse(channel.isActive());
+ assertNull(channel.readOutbound());
+ }
+
+ @Test
+ public void receivedDropsUnexpectedMessageType() throws Exception {
+ EmbeddedChannel channel = new EmbeddedChannel();
+
+ server.received(ControlMessage.CLOSE_MESSAGE, "remote", channel);
+
+ assertFalse(channel.isActive());
+ assertNull(channel.readOutbound());
+ }
+
+ @Test
+ public void undecodableFrameClosesConnection() {
+ EmbeddedChannel channel = pipeline();
+
+ channel.writeInbound(frame(new byte[]{ 0x7f, 0x7f, 0x7f, 0x7f }));
+
+ assertFalse(channel.isActive());
+ }
+
+ @Test
+ public void requestFailureClosesConnection() {
+ EmbeddedChannel channel = pipeline();
+
+ channel.writeInbound(frame(new
HBMessage(HBServerMessageType.SEND_PULSE, null)));
+
+ assertFalse(channel.isActive());
+ assertNull(channel.readOutbound());
+ }
+
+ @Test
+ public void digestHandshakeAuthenticatesChannel() throws Exception {
+ ISaslServer saslServer = mock(ISaslServer.class);
+ when(saslServer.name()).thenReturn("pacemaker_server");
+ when(saslServer.secretKey()).thenReturn("secret");
+ ISaslClient saslClient = mock(ISaslClient.class);
+ when(saslClient.name()).thenReturn("pacemaker_server");
+ when(saslClient.secretKey()).thenReturn("secret");
+
+ EmbeddedChannel serverChannel = new EmbeddedChannel(
+ new ThriftNettyServerCodec(saslServer, config(),
AuthMethod.DIGEST, MAX_LENGTH));
+ EmbeddedChannel clientChannel = new EmbeddedChannel(
+ new ThriftEncoder(), new ThriftDecoder(MAX_LENGTH), new
SaslStormClientHandler(saslClient));
+
+ exchange(clientChannel, serverChannel);
+
+ verify(saslServer).authenticated(serverChannel);
+ verify(saslClient, atLeastOnce()).channelReady(any(Channel.class));
+
assertNull(serverChannel.pipeline().get(ThriftNettyServerCodec.SASL_HANDLER));
+ assertTrue(serverChannel.isActive());
+ assertTrue(clientChannel.isActive());
+ verify(saslServer, never()).received(any(), anyString(),
any(Channel.class));
+
+ HBMessage request = new HBMessage(HBServerMessageType.GET_PULSE,
HBMessageData.path("/path"));
+ serverChannel.writeInbound(frame(request));
+ verify(saslServer).received(request,
serverChannel.remoteAddress().toString(), serverChannel);
+
+ // Control frames other than the handshake request are still refused
once authenticated.
+ serverChannel.writeInbound(controlFrame(ControlMessage.EOB_MESSAGE));
+ assertFalse(serverChannel.isActive());
+ }
+
+ private static void exchange(EmbeddedChannel client, EmbeddedChannel
server) {
+ boolean progress = true;
+ while (progress) {
+ progress = forward(client, server) | forward(server, client);
+ }
+ }
+
+ private static boolean forward(EmbeddedChannel from, EmbeddedChannel to) {
+ List<Object> frames = new ArrayList<>();
+ Object out;
+ while ((out = from.readOutbound()) != null) {
+ frames.add(out);
+ }
+ for (Object frame : frames) {
+ to.writeInbound(frame);
+ }
+ return !frames.isEmpty();
+ }
+}
