Storm拓扑在OptionalDataException引起的StormServerHandler Netty错误后停止发送。

在Storm拓扑中，如果出现OptionalDataException引起的StormServerHandler Netty错误，可以通过以下方法停止发送：

1. 在Storm拓扑的代码中捕获OptionalDataException异常，并在捕获到异常后停止发送数据。

public class MyBolt extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        try {
            // 处理数据
            // ...
            
            // 发送数据
            collector.emit(input, new Values(data));
            collector.ack(input);
        } catch (OptionalDataException e) {
            // 异常处理逻辑
            collector.fail(input);
            // 停止发送数据
            collector.emit("control_stream", new Values("stop"));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("control_stream", new Fields("command"));
        declarer.declare(new Fields("data"));
    }
}

在上面的代码中，通过捕获OptionalDataException异常，我们停止了发送数据，并通过另外一个stream（"control_stream"）发送了一个控制命令（"stop"）。

2. 在拓扑的其它组件中接收这个控制命令，并在接收到停止命令后停止发送数据。

public class MyBolt2 extends BaseRichBolt {
    private OutputCollector collector;

    @Override
    public void prepare(Map<String, Object> conf, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
    }

    @Override
    public void execute(Tuple input) {
        if (input.getSourceStreamId().equals("control_stream")) {
            String command = input.getStringByField("command");
            if (command.equals("stop")) {
                // 停止发送数据
                collector.emit("control_stream", new Values("stop"));
            }
        } else {
            // 处理数据
            // ...
            
            // 发送数据
            collector.emit(input, new Values(data));
            collector.ack(input);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("control_stream", new Fields("command"));
        declarer.declare(new Fields("data"));
    }
}

在上面的代码中，我们在另外一个bolt中接收到了控制命令（"stop"），然后停止了发送数据。

这样，当OptionalDataException发生时，拓扑中的组件将停止发送数据，并知会其它组件停止发送。