You are viewing a plain text version of this content. The canonical link for it is here.

Posted to user@storm.apache.org by kalliopi kalantzaki <ts...@yahoo.gr> on 2017/07/21 06:29:06 UTC

Σχετ: Storm Configuration Serialization problem storm 1.1.0

Thank you for your response. 
Yes, the problem is when i add to configuration the object AESEncryption. However, even if i comment all private members of this class and just use an empty class object, the problem remains. Even if i just use conf.registerSerialization(AESEncryption.class) without the custom serializer the problem is the same. Which means that storm does not get any serialization on non-primitive types in configuration. See the code...
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();
conf.put("aes",new AESEncryption());
cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

   // public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
   // private SecretKey privateKey;.
   // private IvParameterSpec entrophy;
   // private String encryptedMsg;
   // public static final int bitLength = 256;
//.. setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}



    Στις 5:09 μ.μ. Πέμπτη, 20 Ιουλίου 2017, ο/η Bobby Evans <ev...@yahoo-inc.com> έγραψε:
 

 FYI I filed https://issues.apache.org/jira/browse/STORM-2649 to make the exception better.  It should be fairly simple to do (aka I marked it as a newbie jira) if someone wants to look into try and take a crack at it.


- Bobby


On Thursday, July 20, 2017, 8:49:20 AM CDT, Bobby Evans <ev...@yahoo-inc.com> wrote:

I am not sure which config is causing the issue.  You can start testing it by calling
assert org.apache.storm.utils.Utils.isValidConf(conf);
as you add things to the config to find which config is causing the issues.  Sadly the code does not provide any information about what it didn't like about the config.
You can look at the code for the validity check

https://github.com/apache/storm/blob/v1.1.0/storm-core/src/jvm/org/apache/storm/utils/Utils.java#L476-L502
It should not be too hard to modify it to at least log which entries are different when it does the check.

- Bobby


On Thursday, July 20, 2017, 2:44:48 AM CDT, kalliopi kalantzaki <ts...@yahoo.gr> wrote:

Hello,
I am trying to pass to storm configuration an object (not priminitive type), the AESEncryption.class. I created a custom kryo serializer the AESSerializer class and in main topology i used: 
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

    public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
    private SecretKey privateKey;.
    private IvParameterSpec entrophy;
    private String encryptedMsg;
    public static final int bitLength = 256;
... setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}
However, even using kryo serialization i always get the same exception....Problem initializing cluster topology:
java.lang.IllegalArgumentException: Topology conf is not json-serializable
    at org.apache.storm.testing$submit_local_topology.invoke(testing.clj:308) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:49) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-1.1.0.jar:1.1.0]
    at com.jrtechnologies.ifg.bsp.kafka.storm.StormKafkaStarter.main(StormKafkaStarter.java:117) [classes/:?]

Any ideas what am i doing wrong?

Re: Σχετ: Storm Configuration Serialization problem storm 1.1.0

Posted by Bobby Evans <ev...@yahoo-inc.com>.

The issue is that storm does not use kryo serialization for a topology conf.  It uses JSON, and JSON does not support arbitrary objects, just string, map, list, boolean, and numbers.

Your "aes" config cannot be an arbitrary object, it needs to conform to the above types.

- Bobby


On Friday, July 21, 2017, 2:07:06 AM CDT, kalliopi kalantzaki <ts...@yahoo.gr> wrote:

Also, i run the code in Utils, the problem is in line (Map) JSONValue.parse(JSONValue.toJSONString(stormConf)) in isValidConf, the JSONValue.parse() returns a null object. 
 
 

    Στις 9:33 π.μ. Παρασκευή, 21 Ιουλίου 2017, ο/η kalliopi kalantzaki <ts...@yahoo.gr> έγραψε:
 

 Thank you for your response. 
Yes, the problem is when i add to configuration the object AESEncryption. However, even if i comment all private members of this class and just use an empty class object, the problem remains. Even if i just use conf.registerSerialization(AESEncryption.class) without the custom serializer the problem is the same. Which means that storm does not get any serialization on non-primitive types in configuration. See the code...
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();
conf.put("aes",new AESEncryption());
cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

   // public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
   // private SecretKey privateKey;.
   // private IvParameterSpec entrophy;
   // private String encryptedMsg;
   // public static final int bitLength = 256;
//.. setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}



    Στις 5:09 μ.μ. Πέμπτη, 20 Ιουλίου 2017, ο/η Bobby Evans <ev...@yahoo-inc.com> έγραψε:
 

 FYI I filed https://issues.apache.org/jira/browse/STORM-2649 to make the exception better.  It should be fairly simple to do (aka I marked it as a newbie jira) if someone wants to look into try and take a crack at it.


- Bobby


On Thursday, July 20, 2017, 8:49:20 AM CDT, Bobby Evans <ev...@yahoo-inc.com> wrote:

I am not sure which config is causing the issue.  You can start testing it by calling
assert org.apache.storm.utils.Utils.isValidConf(conf);
as you add things to the config to find which config is causing the issues.  Sadly the code does not provide any information about what it didn't like about the config.
You can look at the code for the validity check

https://github.com/apache/storm/blob/v1.1.0/storm-core/src/jvm/org/apache/storm/utils/Utils.java#L476-L502
It should not be too hard to modify it to at least log which entries are different when it does the check.

- Bobby


On Thursday, July 20, 2017, 2:44:48 AM CDT, kalliopi kalantzaki <ts...@yahoo.gr> wrote:

Hello,
I am trying to pass to storm configuration an object (not priminitive type), the AESEncryption.class. I created a custom kryo serializer the AESSerializer class and in main topology i used: 
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

    public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
    private SecretKey privateKey;.
    private IvParameterSpec entrophy;
    private String encryptedMsg;
    public static final int bitLength = 256;
... setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}
However, even using kryo serialization i always get the same exception....Problem initializing cluster topology:
java.lang.IllegalArgumentException: Topology conf is not json-serializable
    at org.apache.storm.testing$submit_local_topology.invoke(testing.clj:308) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:49) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-1.1.0.jar:1.1.0]
    at com.jrtechnologies.ifg.bsp.kafka.storm.StormKafkaStarter.main(StormKafkaStarter.java:117) [classes/:?]

Any ideas what am i doing wrong?

Σχετ: Storm Configuration Serialization problem storm 1.1.0

Posted by kalliopi kalantzaki <ts...@yahoo.gr>.

Also, i run the code in Utils, the problem is in line (Map) JSONValue.parse(JSONValue.toJSONString(stormConf)) in isValidConf, the JSONValue.parse() returns a null object. 
 
 

    Στις 9:33 π.μ. Παρασκευή, 21 Ιουλίου 2017, ο/η kalliopi kalantzaki <ts...@yahoo.gr> έγραψε:
 

 Thank you for your response. 
Yes, the problem is when i add to configuration the object AESEncryption. However, even if i comment all private members of this class and just use an empty class object, the problem remains. Even if i just use conf.registerSerialization(AESEncryption.class) without the custom serializer the problem is the same. Which means that storm does not get any serialization on non-primitive types in configuration. See the code...
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();
conf.put("aes",new AESEncryption());
cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

   // public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
   // private SecretKey privateKey;.
   // private IvParameterSpec entrophy;
   // private String encryptedMsg;
   // public static final int bitLength = 256;
//.. setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}



    Στις 5:09 μ.μ. Πέμπτη, 20 Ιουλίου 2017, ο/η Bobby Evans <ev...@yahoo-inc.com> έγραψε:
 

 FYI I filed https://issues.apache.org/jira/browse/STORM-2649 to make the exception better.  It should be fairly simple to do (aka I marked it as a newbie jira) if someone wants to look into try and take a crack at it.


- Bobby


On Thursday, July 20, 2017, 8:49:20 AM CDT, Bobby Evans <ev...@yahoo-inc.com> wrote:

I am not sure which config is causing the issue.  You can start testing it by calling
assert org.apache.storm.utils.Utils.isValidConf(conf);
as you add things to the config to find which config is causing the issues.  Sadly the code does not provide any information about what it didn't like about the config.
You can look at the code for the validity check

https://github.com/apache/storm/blob/v1.1.0/storm-core/src/jvm/org/apache/storm/utils/Utils.java#L476-L502
It should not be too hard to modify it to at least log which entries are different when it does the check.

- Bobby


On Thursday, July 20, 2017, 2:44:48 AM CDT, kalliopi kalantzaki <ts...@yahoo.gr> wrote:

Hello,
I am trying to pass to storm configuration an object (not priminitive type), the AESEncryption.class. I created a custom kryo serializer the AESSerializer class and in main topology i used: 
Config conf=new Config();conf.registerSerialization(AESEncryption.class, AESSerializer.class);conf.setFallBackOnJavaSerialization(false);
LocalCluster cluster = new LocalCluster();cluster.submitTopology("stormKafkaTopology", conf, builder.createTopology());
public class AESEncryption{

    public static final String cypherAlgorithm = "AES/CBC/PKCS5Padding";
    private SecretKey privateKey;.
    private IvParameterSpec entrophy;
    private String encryptedMsg;
    public static final int bitLength = 256;
... setters getterss methods}

public class AESSerilizer extends Serializer<AESEncryption> implements Serializable {

    public AESSerilizer() {
    }

    @Override
    public void write(Kryo kryo, Output output, AESEncryption object) {


        kryo.writeClassAndObject(output, object);

    }

    @Override
    public AESEncryption read(Kryo kryo, Input input, Class<AESEncryption> type) {
        return (AESEncryption) kryo.readClassAndObject(input);
    }

}
However, even using kryo serialization i always get the same exception....Problem initializing cluster topology:
java.lang.IllegalArgumentException: Topology conf is not json-serializable
    at org.apache.storm.testing$submit_local_topology.invoke(testing.clj:308) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster$_submitTopology.invoke(LocalCluster.clj:49) ~[storm-core-1.1.0.jar:1.1.0]
    at org.apache.storm.LocalCluster.submitTopology(Unknown Source) ~[storm-core-1.1.0.jar:1.1.0]
    at com.jrtechnologies.ifg.bsp.kafka.storm.StormKafkaStarter.main(StormKafkaStarter.java:117) [classes/:?]

Any ideas what am i doing wrong?