如何更改默认的 kafka SpoutConfig 类

Question

我从 kafka 主题获取 3MB 的消息流，但默认值为 1MB。现在我通过在 kafa consumer.properties 和 server.properties 文件中添加以下行，将 kafka 属性从 1MB 更改为 3MB。

fetch.message.max.bytes=2048576 ( consumer.properties )
filemessage.max.bytes=2048576 ( server.properties )
replica.fetch.max.bytes=2048576 ( server.properties )

现在在 Kafka 中添加上述行后，3MB 消息数据将进入 kafka 数据日志。但 STORM 无法处理 3MB 数据，它只能读取默认大小，即 1MB 数据。

那么如何更改这些配置以处理/读取 3MB 数据。这是我的拓扑类。

    String argument = args[0];
    Config conf = new Config();
    conf.put(JDBC_CONF, map);
    conf.setDebug(true);
    conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
    //set the number of workers
    conf.setNumWorkers(3);

    TopologyBuilder builder = new TopologyBuilder();

    //Setup Kafka spout
    BrokerHosts hosts = new ZkHosts("localhost:2181");
    String topic = "year1234"; 
    String zkRoot = "";
    String consumerGroupId = "group1";
    SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);

        spoutConfig.scheme = new SchemeAsMultiScheme(new StringScheme());

    KafkaSpout kafkaSpout = new KafkaSpout(spoutConfig);
    builder.setSpout("KafkaSpout", kafkaSpout,1);


    builder.setBolt("user_details", new Parserspout(),1).shuffleGrouping("KafkaSpout");

    builder.setBolt("bolts_user", new bolts_user(cp),1).shuffleGrouping("user_details");

Answer 1

在下面添加以下行

 SpoutConfig spoutConfig = new SpoutConfig(hosts, topic, zkRoot, consumerGroupId);

spoutConfig.fetchSizeBytes = 3048576;
spoutConfig.bufferSizeBytes = 3048576;