You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by xi...@apache.org on 2017/07/20 14:10:03 UTC
[1/5] storm git commit: Adding in maximum/minimum object size
Repository: storm
Updated Branches:
refs/heads/master 8814bfaa1 -> 5d93a8644
Adding in maximum/minimum object size
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/e6465b5d
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/e6465b5d
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/e6465b5d
Branch: refs/heads/master
Commit: e6465b5dc5c7e31e0b92a18de578a13d05a5abf2
Parents: 559053d
Author: Kyle Nusbaum <kn...@yahoo-inc.com>
Authored: Mon Jul 17 14:40:15 2017 -0500
Committer: Kyle Nusbaum <kn...@yahoo-inc.com>
Committed: Mon Jul 17 14:40:15 2017 -0500
----------------------------------------------------------------------
conf/defaults.yaml | 1 +
storm-client/src/jvm/org/apache/storm/Config.java | 7 +++++++
.../org/apache/storm/pacemaker/PacemakerClient.java | 3 ++-
.../apache/storm/pacemaker/codec/ThriftDecoder.java | 15 +++++++++++++++
.../pacemaker/codec/ThriftNettyClientCodec.java | 6 ++++--
.../org/apache/storm/pacemaker/PacemakerServer.java | 4 ++--
.../pacemaker/codec/ThriftNettyServerCodec.java | 5 ++---
7 files changed, 33 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/conf/defaults.yaml
----------------------------------------------------------------------
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index eb9cb24..0078323 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -282,6 +282,7 @@ pacemaker.thread.timeout: 10
pacemaker.childopts: "-Xmx1024m"
pacemaker.auth.method: "NONE"
pacemaker.kerberos.users: []
+pacemaker.thrift.message.size.max: 10485760
#default storm daemon metrics reporter plugins
storm.daemon.metrics.reporter.plugins:
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index 5d5f723..367e7ed 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -804,6 +804,13 @@ public class Config extends HashMap<String, Object> {
public static final String PACEMAKER_AUTH_METHOD = "pacemaker.auth.method";
/**
+ * Pacemaker Thrift Max Message Size (bytes)
+ */
+ @isInteger
+ @isPositiveNumber
+ 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
*/
@isInteger
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
----------------------------------------------------------------------
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 be09b3e..bdfdeae 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -117,7 +117,8 @@ public class PacemakerClient implements ISaslClient {
bootstrap.setOption("keepAlive", true);
remote_addr = new InetSocketAddress(host, port);
- ChannelPipelineFactory pipelineFactory = new ThriftNettyClientCodec(this, config, authMethod, host).pipelineFactory();
+ int thriftMessageMaxSize = (Integer)config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
+ ChannelPipelineFactory pipelineFactory = new ThriftNettyClientCodec(this, config, authMethod, host, thriftMessageMaxSize).pipelineFactory();
bootstrap.setPipelineFactory(pipelineFactory);
bootstrap.connect(remote_addr);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
----------------------------------------------------------------------
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 968ca34..58d2b1b 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
@@ -27,10 +27,18 @@ import org.jboss.netty.buffer.ChannelBuffer;
import org.apache.storm.messaging.netty.ControlMessage;
import org.apache.storm.messaging.netty.SaslMessageToken;
+import java.io.IOException;
+
public class ThriftDecoder extends FrameDecoder {
private static final int INTEGER_SIZE = 4;
+ private int maxLength;
+
+ public ThriftDecoder(int maxLength) {
+ this.maxLength = maxLength;
+ }
+
@Override
protected Object decode(ChannelHandlerContext ctx, Channel channel, ChannelBuffer buf) throws Exception {
@@ -42,6 +50,13 @@ public class ThriftDecoder extends FrameDecoder {
buf.markReaderIndex();
int thriftLen = buf.readInt();
+
+ if(thriftLen < 0 || thriftLen > maxLength) {
+ throw new IOException("Thrift message of length " + Integer.toString(thriftLen)
+ + " is greater than allowed " + maxLength
+ + " or less than 0.");
+ }
+
available -= INTEGER_SIZE;
if(available < thriftLen) {
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
----------------------------------------------------------------------
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 42a68fd..1ad5a6c 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
@@ -48,12 +48,14 @@ public class ThriftNettyClientCodec {
private AuthMethod authMethod;
private Map<String, Object> topoConf;
private String host;
+ private int thriftMessageMaxSize;
- public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf, AuthMethod authMethod, String host) {
+ public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf, AuthMethod authMethod, String host, int thriftMessageMaxSize) {
client = pacemaker_client;
this.authMethod = authMethod;
this.topoConf = topoConf;
this.host = host;
+ this.thriftMessageMaxSize = thriftMessageMaxSize;
}
public ChannelPipelineFactory pipelineFactory() {
@@ -61,7 +63,7 @@ public class ThriftNettyClientCodec {
public ChannelPipeline getPipeline() {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("encoder", new ThriftEncoder());
- pipeline.addLast("decoder", new ThriftDecoder());
+ pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSize));
if (authMethod == AuthMethod.KERBEROS) {
try {
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
----------------------------------------------------------------------
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 315d0ac..5a68ac8 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
@@ -106,8 +106,8 @@ class PacemakerServer implements ISaslServer {
bootstrap.setOption("tcpNoDelay", true);
bootstrap.setOption("sendBufferSize", FIVE_MB_IN_BYTES);
bootstrap.setOption("keepAlive", true);
-
- ChannelPipelineFactory pipelineFactory = new ThriftNettyServerCodec(this, config, authMethod).pipelineFactory();
+ int thriftMessageMaxSize = (Integer)config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
+ ChannelPipelineFactory pipelineFactory = new ThriftNettyServerCodec(this, config, authMethod, thriftMessageMaxSize).pipelineFactory();
bootstrap.setPipelineFactory(pipelineFactory);
Channel channel = bootstrap.bind(new InetSocketAddress(port));
allChannels.add(channel);
http://git-wip-us.apache.org/repos/asf/storm/blob/e6465b5d/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
----------------------------------------------------------------------
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 8b0de60..0dc6338 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
@@ -17,7 +17,6 @@
*/
package org.apache.storm.pacemaker.codec;
-import org.apache.storm.Config;
import org.apache.storm.DaemonConfig;
import org.apache.storm.messaging.netty.ISaslServer;
import org.apache.storm.messaging.netty.IServer;
@@ -52,7 +51,7 @@ public class ThriftNettyServerCodec {
private static final Logger LOG = LoggerFactory
.getLogger(ThriftNettyServerCodec.class);
- public ThriftNettyServerCodec(IServer server, Map<String, Object> topoConf, AuthMethod authMethod) {
+ public ThriftNettyServerCodec(IServer server, Map<String, Object> topoConf, AuthMethod authMethod, int thriftMessageMaxSize) {
this.server = server;
this.authMethod = authMethod;
this.topoConf = topoConf;
@@ -64,7 +63,7 @@ public class ThriftNettyServerCodec {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("encoder", new ThriftEncoder());
- pipeline.addLast("decoder", new ThriftDecoder());
+ pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSize));
if(authMethod == AuthMethod.DIGEST) {
try {
LOG.debug("Adding SaslStormServerHandler to pacemaker server pipeline.");
[4/5] storm git commit: More checkstyle cleaning.
Posted by xi...@apache.org.
More checkstyle cleaning.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/a9611ef7
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/a9611ef7
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/a9611ef7
Branch: refs/heads/master
Commit: a9611ef74a905cfe0f60cade12c55fb3fe6269d8
Parents: 9f621e5
Author: Kyle Nusbaum <kn...@yahoo-inc.com>
Authored: Wed Jul 19 12:04:09 2017 -0500
Committer: Kyle Nusbaum <kn...@yahoo-inc.com>
Committed: Wed Jul 19 12:04:09 2017 -0500
----------------------------------------------------------------------
.../org/apache/storm/pacemaker/PacemakerClient.java | 6 ++----
.../apache/storm/pacemaker/codec/ThriftDecoder.java | 15 +++++++++++----
.../pacemaker/codec/ThriftNettyClientCodec.java | 9 +++++----
.../org/apache/storm/pacemaker/PacemakerServer.java | 4 +++-
.../pacemaker/codec/ThriftNettyServerCodec.java | 7 ++++---
5 files changed, 25 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/a9611ef7/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
----------------------------------------------------------------------
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 33535a2..c43bd16 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -117,11 +117,9 @@ public class PacemakerClient implements ISaslClient {
bootstrap.setOption("keepAlive", true);
remote_addr = new InetSocketAddress(host, port);
- int thriftMessageMaxSize =
- (Integer) config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
+ int thriftMessageMaxSize = (Integer) config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
ChannelPipelineFactory pipelineFactory =
- new ThriftNettyClientCodec(this, config, authMethod,
- host, thriftMessageMaxSize)
+ new ThriftNettyClientCodec(this, config, authMethod, host, thriftMessageMaxSize)
.pipelineFactory();
bootstrap.setPipelineFactory(pipelineFactory);
bootstrap.connect(remote_addr);
http://git-wip-us.apache.org/repos/asf/storm/blob/a9611ef7/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
----------------------------------------------------------------------
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 c9d7bc3..9e37ffb 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
@@ -32,10 +32,18 @@ public class ThriftDecoder extends FrameDecoder {
private static final int INTEGER_SIZE = 4;
- private int maxLength;
+ /**
+ * The maximum length in bytes that a serialized thrift message is allowed
+ * to be.
+ */
+ private final int maxLength;
- public ThriftDecoder(int maxLength) {
- this.maxLength = maxLength;
+ /**
+ * Instantiate a ThriftDecoder that accepts serialized messages of at most
+ * maxLength bytes.
+ */
+ public ThriftDecoder(final int maxLengthBytes) {
+ maxLength = maxLengthBytes;
}
@Override
@@ -49,7 +57,6 @@ public class ThriftDecoder extends FrameDecoder {
buf.markReaderIndex();
int thriftLen = buf.readInt();
-
if (thriftLen < 0 || thriftLen > maxLength) {
throw new IOException("Thrift message of length " + Integer.toString(thriftLen)
+ " is greater than allowed " + maxLength
http://git-wip-us.apache.org/repos/asf/storm/blob/a9611ef7/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
----------------------------------------------------------------------
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 dbe4a5d..811a72f 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
@@ -48,14 +48,15 @@ public class ThriftNettyClientCodec {
private AuthMethod authMethod;
private Map<String, Object> topoConf;
private String host;
- private final int _thriftMessageMaxSize;
+ private final int thriftMessageMaxSize;
- public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf, AuthMethod authMethod, String host, int thriftMessageMaxSize) {
+ public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf,
+ AuthMethod authMethod, String host, int thriftMessageMaxSizeBytes) {
client = pacemaker_client;
this.authMethod = authMethod;
this.topoConf = topoConf;
this.host = host;
- _thriftMessageMaxSize = thriftMessageMaxSize;
+ thriftMessageMaxSize = thriftMessageMaxSizeBytes;
}
public ChannelPipelineFactory pipelineFactory() {
@@ -63,7 +64,7 @@ public class ThriftNettyClientCodec {
public ChannelPipeline getPipeline() {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("encoder", new ThriftEncoder());
- pipeline.addLast("decoder", new ThriftDecoder(_thriftMessageMaxSize));
+ pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSize));
if (authMethod == AuthMethod.KERBEROS) {
try {
http://git-wip-us.apache.org/repos/asf/storm/blob/a9611ef7/storm-server/src/main/java/org/apache/storm/pacemaker/PacemakerServer.java
----------------------------------------------------------------------
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 5a68ac8..2f050db 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
@@ -107,7 +107,9 @@ class PacemakerServer implements ISaslServer {
bootstrap.setOption("sendBufferSize", FIVE_MB_IN_BYTES);
bootstrap.setOption("keepAlive", true);
int thriftMessageMaxSize = (Integer)config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
- ChannelPipelineFactory pipelineFactory = new ThriftNettyServerCodec(this, config, authMethod, thriftMessageMaxSize).pipelineFactory();
+ ChannelPipelineFactory pipelineFactory =
+ new ThriftNettyServerCodec(this, config, authMethod, thriftMessageMaxSize)
+ .pipelineFactory();
bootstrap.setPipelineFactory(pipelineFactory);
Channel channel = bootstrap.bind(new InetSocketAddress(port));
allChannels.add(channel);
http://git-wip-us.apache.org/repos/asf/storm/blob/a9611ef7/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
----------------------------------------------------------------------
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 80e9c10..a6ceade 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
@@ -47,16 +47,17 @@ public class ThriftNettyServerCodec {
private IServer server;
private AuthMethod authMethod;
private Map<String, Object> topoConf;
- private int thriftMessageMaxSize;
+ private final int thriftMessageMaxSize;
private static final Logger LOG = LoggerFactory
.getLogger(ThriftNettyServerCodec.class);
- public ThriftNettyServerCodec(IServer server, Map<String, Object> topoConf, AuthMethod authMethod, int thriftMessageMaxSize) {
+ public ThriftNettyServerCodec(IServer server, Map<String, Object> topoConf,
+ AuthMethod authMethod, int thriftMessageMaxSizeBytes) {
this.server = server;
this.authMethod = authMethod;
this.topoConf = topoConf;
- this.thriftMessageMaxSize = thriftMessageMaxSize;
+ thriftMessageMaxSize = thriftMessageMaxSizeBytes;
}
public ChannelPipelineFactory pipelineFactory() {
[3/5] storm git commit: Fixing some checkstyle complaints.
Posted by xi...@apache.org.
Fixing some checkstyle complaints.
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/9f621e5f
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/9f621e5f
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/9f621e5f
Branch: refs/heads/master
Commit: 9f621e5faeb7ed6129e6032c28a91bead7a11169
Parents: d484779
Author: Kyle Nusbaum <kn...@yahoo-inc.com>
Authored: Wed Jul 19 11:16:12 2017 -0500
Committer: Kyle Nusbaum <kn...@yahoo-inc.com>
Committed: Wed Jul 19 11:16:12 2017 -0500
----------------------------------------------------------------------
storm-client/src/jvm/org/apache/storm/Config.java | 2 +-
.../org/apache/storm/pacemaker/PacemakerClient.java | 8 ++++++--
.../apache/storm/pacemaker/codec/ThriftDecoder.java | 15 +++++++--------
.../pacemaker/codec/ThriftNettyClientCodec.java | 6 +++---
4 files changed, 17 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/9f621e5f/storm-client/src/jvm/org/apache/storm/Config.java
----------------------------------------------------------------------
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java
index 367e7ed..7f1c89e 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -804,7 +804,7 @@ public class Config extends HashMap<String, Object> {
public static final String PACEMAKER_AUTH_METHOD = "pacemaker.auth.method";
/**
- * Pacemaker Thrift Max Message Size (bytes)
+ * Pacemaker Thrift Max Message Size (bytes).
*/
@isInteger
@isPositiveNumber
http://git-wip-us.apache.org/repos/asf/storm/blob/9f621e5f/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
----------------------------------------------------------------------
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 bdfdeae..33535a2 100644
--- a/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
+++ b/storm-client/src/jvm/org/apache/storm/pacemaker/PacemakerClient.java
@@ -117,8 +117,12 @@ public class PacemakerClient implements ISaslClient {
bootstrap.setOption("keepAlive", true);
remote_addr = new InetSocketAddress(host, port);
- int thriftMessageMaxSize = (Integer)config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
- ChannelPipelineFactory pipelineFactory = new ThriftNettyClientCodec(this, config, authMethod, host, thriftMessageMaxSize).pipelineFactory();
+ int thriftMessageMaxSize =
+ (Integer) config.get(Config.PACEMAKER_THRIFT_MESSAGE_SIZE_MAX);
+ ChannelPipelineFactory pipelineFactory =
+ new ThriftNettyClientCodec(this, config, authMethod,
+ host, thriftMessageMaxSize)
+ .pipelineFactory();
bootstrap.setPipelineFactory(pipelineFactory);
bootstrap.connect(remote_addr);
}
http://git-wip-us.apache.org/repos/asf/storm/blob/9f621e5f/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftDecoder.java
----------------------------------------------------------------------
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 58d2b1b..c9d7bc3 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
@@ -17,17 +17,16 @@
*/
package org.apache.storm.pacemaker.codec;
-import org.apache.storm.utils.Utils;
-import org.jboss.netty.handler.codec.frame.FrameDecoder;
-import org.jboss.netty.channel.ChannelHandlerContext;
-import org.jboss.netty.channel.Channel;
+import java.io.IOException;
import org.apache.storm.generated.HBMessage;
import org.apache.storm.generated.HBServerMessageType;
-import org.jboss.netty.buffer.ChannelBuffer;
import org.apache.storm.messaging.netty.ControlMessage;
import org.apache.storm.messaging.netty.SaslMessageToken;
-
-import java.io.IOException;
+import org.apache.storm.utils.Utils;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.handler.codec.frame.FrameDecoder;
public class ThriftDecoder extends FrameDecoder {
@@ -51,7 +50,7 @@ public class ThriftDecoder extends FrameDecoder {
int thriftLen = buf.readInt();
- if(thriftLen < 0 || thriftLen > maxLength) {
+ if (thriftLen < 0 || thriftLen > maxLength) {
throw new IOException("Thrift message of length " + Integer.toString(thriftLen)
+ " is greater than allowed " + maxLength
+ " or less than 0.");
http://git-wip-us.apache.org/repos/asf/storm/blob/9f621e5f/storm-client/src/jvm/org/apache/storm/pacemaker/codec/ThriftNettyClientCodec.java
----------------------------------------------------------------------
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 1ad5a6c..dbe4a5d 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
@@ -48,14 +48,14 @@ public class ThriftNettyClientCodec {
private AuthMethod authMethod;
private Map<String, Object> topoConf;
private String host;
- private int thriftMessageMaxSize;
+ private final int _thriftMessageMaxSize;
public ThriftNettyClientCodec(PacemakerClient pacemaker_client, Map<String, Object> topoConf, AuthMethod authMethod, String host, int thriftMessageMaxSize) {
client = pacemaker_client;
this.authMethod = authMethod;
this.topoConf = topoConf;
this.host = host;
- this.thriftMessageMaxSize = thriftMessageMaxSize;
+ _thriftMessageMaxSize = thriftMessageMaxSize;
}
public ChannelPipelineFactory pipelineFactory() {
@@ -63,7 +63,7 @@ public class ThriftNettyClientCodec {
public ChannelPipeline getPipeline() {
ChannelPipeline pipeline = Channels.pipeline();
pipeline.addLast("encoder", new ThriftEncoder());
- pipeline.addLast("decoder", new ThriftDecoder(thriftMessageMaxSize));
+ pipeline.addLast("decoder", new ThriftDecoder(_thriftMessageMaxSize));
if (authMethod == AuthMethod.KERBEROS) {
try {
[2/5] storm git commit: Codec fix
Posted by xi...@apache.org.
Codec fix
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/d4847797
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/d4847797
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/d4847797
Branch: refs/heads/master
Commit: d4847797f6ad1d4bd18ab1b79ceccd36c906f1e7
Parents: e6465b5
Author: Kyle Nusbaum <kn...@yahoo-inc.com>
Authored: Mon Jul 17 14:59:45 2017 -0500
Committer: Kyle Nusbaum <kn...@yahoo-inc.com>
Committed: Mon Jul 17 14:59:45 2017 -0500
----------------------------------------------------------------------
.../org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/storm/blob/d4847797/storm-server/src/main/java/org/apache/storm/pacemaker/codec/ThriftNettyServerCodec.java
----------------------------------------------------------------------
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 0dc6338..80e9c10 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
@@ -47,6 +47,7 @@ public class ThriftNettyServerCodec {
private IServer server;
private AuthMethod authMethod;
private Map<String, Object> topoConf;
+ private int thriftMessageMaxSize;
private static final Logger LOG = LoggerFactory
.getLogger(ThriftNettyServerCodec.class);
@@ -55,6 +56,7 @@ public class ThriftNettyServerCodec {
this.server = server;
this.authMethod = authMethod;
this.topoConf = topoConf;
+ this.thriftMessageMaxSize = thriftMessageMaxSize;
}
public ChannelPipelineFactory pipelineFactory() {
[5/5] storm git commit: Merge branch 'STORM-2638' of
https://github.com/knusbaum/incubator-storm
Posted by xi...@apache.org.
Merge branch 'STORM-2638' of https://github.com/knusbaum/incubator-storm
Project: http://git-wip-us.apache.org/repos/asf/storm/repo
Commit: http://git-wip-us.apache.org/repos/asf/storm/commit/5d93a864
Tree: http://git-wip-us.apache.org/repos/asf/storm/tree/5d93a864
Diff: http://git-wip-us.apache.org/repos/asf/storm/diff/5d93a864
Branch: refs/heads/master
Commit: 5d93a864459b2407fda95de4995ac316ddd49fe7
Parents: 8814bfa a9611ef
Author: Xin Wang <be...@163.com>
Authored: Thu Jul 20 22:09:25 2017 +0800
Committer: Xin Wang <be...@163.com>
Committed: Thu Jul 20 22:09:25 2017 +0800
----------------------------------------------------------------------
conf/defaults.yaml | 1 +
.../src/jvm/org/apache/storm/Config.java | 7 +++++
.../apache/storm/pacemaker/PacemakerClient.java | 5 +++-
.../storm/pacemaker/codec/ThriftDecoder.java | 31 ++++++++++++++++----
.../pacemaker/codec/ThriftNettyClientCodec.java | 7 +++--
.../apache/storm/pacemaker/PacemakerServer.java | 6 ++--
.../pacemaker/codec/ThriftNettyServerCodec.java | 8 +++--
7 files changed, 52 insertions(+), 13 deletions(-)
----------------------------------------------------------------------