class CollectorOnlineProcessor implements NettyRemotingProcessor {
    private final ManageServer manageServer;

    public CollectorOnlineProcessor(final ManageServer manageServer) {
        this.manageServer = manageServer;
    }

    @Override
    public ClusterMsg.Message handle(ChannelHandlerContext ctx, ClusterMsg.Message message) {<FILL_FUNCTION_BODY>}
}

        String collector = message.getIdentity();
        log.info("the collector {} actively requests to go online.", collector);
        CollectorInfo collectorInfo = JsonUtil.fromJson(message.getMsg(), CollectorInfo.class);
        this.manageServer.addChannel(collector, ctx.channel());
        this.manageServer.getCollectorAndJobScheduler().collectorGoOnline(collector, collectorInfo);
        return null;
    
<no_super_class>