（六）storm-kafka源代码走读之PartitionManager

PartitionManager算是storm-kafka的核心类了，如今開始简单分析一下。还是先声明一下，metric部分这里不做分析。

PartitionManager主要负责的是消息的发送、容错处理，所以PartitionManager会有三个集合

•  _pending：尚未发送的message的offset集合， 是个TreeSet<Long>()
• failed ： 发送失败的offset 集合，是个TreeSet<Long>()
• _waitingToEmit： 存放待发射的message，是个LinkedList。从kafka读到的message所有先放在这里

还有几个变量须要说一下：

• Long _emittedToOffset;   从kafka读到的offset。从kafka读到的messages会放入_waitingToEmit，放入这个list。我们就觉得一定会被emit，所以emittedToOffset能够觉得是从kafka读到的offset 
• Long _committedTo;  已经写入zk的offset
• lastCompletedOffset() 已经被成功处理的offset，因为message是要在storm里面处理的，当中是可能fail的，所以正在处理的offset是缓存在_pending中的 
  假设_pending为空，那么lastCompletedOffset=_emittedToOffset 
  假设_pending不为空，那么lastCompletedOffset为pending list里面第一个offset，由于后面都还在等待ack

public long lastCompletedOffset() {
        if (_pending.isEmpty()) {
            return _emittedToOffset;
        } else {
            return _pending.first();
        }
    }

来看初始化的过程：

     public PartitionManager(DynamicPartitionConnections connections, String topologyInstanceId, ZkState state, Map stormConf, SpoutConfig spoutConfig, Partition id) {
        _partition = id;
        _connections = connections;
        _spoutConfig = spoutConfig;
        _topologyInstanceId = topologyInstanceId;
        _consumer = connections.register(id.host, id.partition);
        _state = state;
        _stormConf = stormConf;
        numberAcked = numberFailed = 0;

        String jsonTopologyId = null;
        Long jsonOffset = null;
        String path = committedPath();
        try {
            Map<Object, Object> json = _state.readJSON(path); // 从zk读取offset
            LOG.info("Read partition information from: " + path +  "  --> " + json );
            if (json != null) {
                jsonTopologyId = (String) ((Map<Object, Object>) json.get("topology")).get("id");
                jsonOffset = (Long) json.get("offset");
            }
        } catch (Throwable e) {
            LOG.warn("Error reading and/or parsing at ZkNode: " + path, e);
        }

		/**
		 * 依据用户设置的startOffsetTime，值来读取offset（-2 从kafka头開始  -1 是从最新的開始 0 =无 从ZK開始）
		 **/
        Long currentOffset = KafkaUtils.getOffset(_consumer, spoutConfig.topic, id.partition, spoutConfig);

        if (jsonTopologyId == null || jsonOffset == null) { // failed to parse JSON?
            _committedTo = currentOffset;
            LOG.info("No partition information found, using configuration to determine offset");
        } else if (!topologyInstanceId.equals(jsonTopologyId) && spoutConfig.forceFromStart) {
            _committedTo = KafkaUtils.getOffset(_consumer, spoutConfig.topic, id.partition, spoutConfig.startOffsetTime);
            LOG.info("Topology change detected and reset from start forced, using configuration to determine offset");
        } else {
            _committedTo = jsonOffset;
            LOG.info("Read last commit offset from zookeeper: " + _committedTo + "; old topology_id: " + jsonTopologyId + " - new topology_id: " + topologyInstanceId );
        }

		/**
		 * 以下这个if推断是假设当前读取的offset值与提交到zk的值不一致，且相差Long.MAX_VALUE,就觉得中间非常大部分msg发射了没有提交，就把这部分所有放弃，避免重发
		 * 令_committedTo = currentOffset， 这个是新修复的bug。之前maxOffsetBehind=100000（好像是这个值，这个太小），bug isuue 是 STORM-399

		 **/
        if (currentOffset - _committedTo > spoutConfig.maxOffsetBehind || _committedTo <= 0) {
            LOG.info("Last commit offset from zookeeper: " + _committedTo);
            _committedTo = currentOffset;
            LOG.info("Commit offset " + _committedTo + " is more than " +
                    spoutConfig.maxOffsetBehind + " behind, resetting to startOffsetTime=" + spoutConfig.startOffsetTime);
        }

        LOG.info("Starting Kafka " + _consumer.host() + ":" + id.partition + " from offset " + _committedTo);
        _emittedToOffset = _committedTo;

        _fetchAPILatencyMax = new CombinedMetric(new MaxMetric());
        _fetchAPILatencyMean = new ReducedMetric(new MeanReducer());
        _fetchAPICallCount = new CountMetric();
        _fetchAPIMessageCount = new CountMetric();
    }

刚開始的时候须要读取message，放到_waitingToEmit中。这是fill的过程，看代码

private void fill() {
        long start = System.nanoTime();
        long offset;
		
		// 首先要推断是否有fail的offset， 假设有的话，在须要从这个offset開始往下去读取message，所以这里有重发的可能
        final boolean had_failed = !failed.isEmpty(); 

        // Are there failed tuples?
 If so, fetch those first.
        if (had_failed) {
            offset = failed.first(); // 取失败的最小的offset值。
        } else {
            offset = _emittedToOffset;
        }

        ByteBufferMessageSet msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset);
        long end = System.nanoTime();
        long millis = (end - start) / 1000000;
        _fetchAPILatencyMax.update(millis);
        _fetchAPILatencyMean.update(millis);
        _fetchAPICallCount.incr();
        if (msgs != null) {
            int numMessages = 0;

            for (MessageAndOffset msg : msgs) {
                final Long cur_offset = msg.offset();
                if (cur_offset < offset) {
                    // Skip any old offsets.
                    continue;
                }
				
				/**
				 * 仅仅要是没有失败的或者失败的set中含有该offset（由于失败msg有非常多，我们仅仅是从最小的offset開始读取msg的）
				 * ，就把这个message放到待发射的list中
				 **/
                if (!had_failed || failed.contains(cur_offset)) {
                    numMessages += 1;
                    _pending.add(cur_offset);
                    _waitingToEmit.add(new MessageAndRealOffset(msg.message(), cur_offset));
                    _emittedToOffset = Math.max(msg.nextOffset(), _emittedToOffset);
                    if (had_failed) { // 假设失败列表中含有该offset，就移除，由于要又一次发射了。
                        failed.remove(cur_offset);
                    }
                }
            }
            _fetchAPIMessageCount.incrBy(numMessages);
        }
    }

=================== github 最新代码======================

把例如以下代码改成

ByteBufferMessageSet msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset);

改动成：

ByteBufferMessageSet msgs = null;
        try {
            msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset);
        } catch (UpdateOffsetException e) {
            _emittedToOffset = KafkaUtils.getOffset(_consumer, _spoutConfig.topic, _partition.partition, _spoutConfig);
            LOG.warn("Using new offset: {}", _emittedToOffset);
            // fetch failed, so don't update the metrics
            return;
        }

====================2014-11-10========================

fetchMessage过程

public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) {
        ByteBufferMessageSet msgs = null;
        String topic = config.topic;
        int partitionId = partition.partition;
        for (int errors = 0; errors < 2 && msgs == null; errors++) { //容忍两次错误
            FetchRequestBuilder builder = new FetchRequestBuilder();
            FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes).
                    clientId(config.clientId).maxWait(config.fetchMaxWait).build();
            FetchResponse fetchResponse;
            try {
                fetchResponse = consumer.fetch(fetchRequest);
            } catch (Exception e) {
                if (e instanceof ConnectException ||
                        e instanceof SocketTimeoutException ||
                        e instanceof IOException ||
                        e instanceof UnresolvedAddressException
                        ) {
                    LOG.warn("Network error when fetching messages:", e);
                    throw new FailedFetchException(e);
                } else {
                    throw new RuntimeException(e);
                }
            }
            if (fetchResponse.hasError()) { // 主要处理offset outofrange的case。通过getOffset从earliest或latest读
                KafkaError error = KafkaError.getError(fetchResponse.errorCode(topic, partitionId));
                if (error.equals(KafkaError.OFFSET_OUT_OF_RANGE) && config.useStartOffsetTimeIfOffsetOutOfRange && errors == 0) {
                    long startOffset = getOffset(consumer, topic, partitionId, config.startOffsetTime);
                    LOG.warn("Got fetch request with offset out of range: [" + offset + "]; " +
                            "retrying with default start offset time from configuration. " +
                            "configured start offset time: [" + config.startOffsetTime + "] offset: [" + startOffset + "]");
                    offset = startOffset;
                } else {
                    String message = "Error fetching data from [" + partition + "] for topic [" + topic + "]: [" + error + "]";
                    LOG.error(message);
                    throw new FailedFetchException(message);
                }
            } else {
                msgs = fetchResponse.messageSet(topic, partitionId);
            }
        }
        return msgs;
    }

====================== github 上关于fetchMessage已有相关修改。修改代码例如以下====================================

public static ByteBufferMessageSet fetchMessages(KafkaConfig config, SimpleConsumer consumer, Partition partition, long offset) throws UpdateOffsetException {
        ByteBufferMessageSet msgs = null;
        String topic = config.topic;
        int partitionId = partition.partition;
        FetchRequestBuilder builder = new FetchRequestBuilder();
        FetchRequest fetchRequest = builder.addFetch(topic, partitionId, offset, config.fetchSizeBytes).
                clientId(config.clientId).maxWait(config.fetchMaxWait).build();
        FetchResponse fetchResponse;
        try {
            fetchResponse = consumer.fetch(fetchRequest);
        } catch (Exception e) {
            if (e instanceof ConnectException ||
                    e instanceof SocketTimeoutException ||
                    e instanceof IOException ||
                    e instanceof UnresolvedAddressException
                    ) {
                LOG.warn("Network error when fetching messages:", e);
                throw new FailedFetchException(e);
            } else {
                throw new RuntimeException(e);
            }
        }
        if (fetchResponse.hasError()) {
            KafkaError error = KafkaError.getError(fetchResponse.errorCode(topic, partitionId));
            if (error.equals(KafkaError.OFFSET_OUT_OF_RANGE) && config.useStartOffsetTimeIfOffsetOutOfRange) {
                LOG.warn("Got fetch request with offset out of range: [" + offset + "]; " +
                        "retrying with default start offset time from configuration. " +
                        "configured start offset time: [" + config.startOffsetTime + "]");
                throw new UpdateOffsetException();
            } else {
                String message = "Error fetching data from [" + partition + "] for topic [" + topic + "]: [" + error + "]";
                LOG.error(message);
                throw new FailedFetchException(message);
            }
        } else {
            msgs = fetchResponse.messageSet(topic, partitionId);
        }
        return msgs;
    }

去掉了三次容错处理。之所以这么改动的原因大家去看STORM-511 isuue。描写叙述的原因例如以下：

With default behaviour (KafkaConfig.useStartOffsetTimeIfOffsetOutOfRange == true) when Kafka returns the error about offset being out of range, storm.kafka.KafkaUtils.fetchMessages tries to fix offset in local scope and retry fetch request. But if there are no more messages appeared under that specified partition it will never update the PartitionManager, but keep sending tons of requests with invalid offset to Kafka broker. On both sides Storm and Kafka logs grow extremely quick during that time.

==========================================2014-11-10加入=====================================================

再来看next方法。这种方法就是KafkaSpout 的nextTuple真正调用的方法

//returns false if it's reached the end of current batch
    public EmitState next(SpoutOutputCollector collector) {
        if (_waitingToEmit.isEmpty()) {
            fill(); // 開始时获取message
        }
        while (true) {
            MessageAndRealOffset toEmit = _waitingToEmit.pollFirst(); //每次读取一条
            if (toEmit == null) {
                return EmitState.NO_EMITTED;
            }
			// 假设忘记了，能够再返回看下自己定义scheme这篇 ： http://blog.csdn.net/wzhg0508/article/details/40874155
            Iterable<List<Object>> tups = KafkaUtils.generateTuples(_spoutConfig, toEmit.msg); 
            if (tups != null) {
                for (List<Object> tup : tups) { //这个地方在讲述自己定义Scheme时。提到了
                    collector.emit(tup, new KafkaMessageId(_partition, toEmit.offset));
                }
                break; // 这里就是每成功发射一天msg，就break掉，返回emitstate给kafkaSpout的nextTuple中做推断和定时commit成功处理的offset到zk
            } else {
                ack(toEmit.offset); // ack 做清除工作
            }
        }
        if (!_waitingToEmit.isEmpty()) {
            return EmitState.EMITTED_MORE_LEFT;
        } else {
            return EmitState.EMITTED_END;
        }
    }

来看ack的操作：

public void ack(Long offset) {
        if (!_pending.isEmpty() && _pending.first() < offset - _spoutConfig.maxOffsetBehind) {
            // Too many things pending!
            _pending.headSet(offset - _spoutConfig.maxOffsetBehind).clear();
        }
        _pending.remove(offset);
        numberAcked++;
    }

接下来看下。commit的操作，这个操作也是在kafkaSpout中调用的

public void commit() {
        long lastCompletedOffset = lastCompletedOffset();
        if (_committedTo != lastCompletedOffset) {
            LOG.debug("Writing last completed offset (" + lastCompletedOffset + ") to ZK for " + _partition + " for topology: " + _topologyInstanceId);
            Map<Object, Object> data = (Map<Object, Object>) ImmutableMap.builder()
                    .put("topology", ImmutableMap.of("id", _topologyInstanceId,
                            "name", _stormConf.get(Config.TOPOLOGY_NAME)))
                    .put("offset", lastCompletedOffset)
                    .put("partition", _partition.partition)
                    .put("broker", ImmutableMap.of("host", _partition.host.host,
                            "port", _partition.host.port))
                    .put("topic", _spoutConfig.topic).build();
            _state.writeJSON(committedPath(), data);

            _committedTo = lastCompletedOffset;
            LOG.debug("Wrote last completed offset (" + lastCompletedOffset + ") to ZK for " + _partition + " for topology: " + _topologyInstanceId);
        } else {
            LOG.debug("No new offset for " + _partition + " for topology: " + _topologyInstanceId);
        }
    }

这个commit非常easy。不用我多做解释了

重点来看下fail处理吧

public void fail(Long offset) {
        if (offset < _emittedToOffset - _spoutConfig.maxOffsetBehind) {
            LOG.info(
                    "Skipping failed tuple at offset=" + offset +
                            " because it's more than maxOffsetBehind=" + _spoutConfig.maxOffsetBehind +
                            " behind _emittedToOffset=" + _emittedToOffset
            );
        } else {
            LOG.debug("failing at offset=" + offset + " with _pending.size()=" + _pending.size() + " pending and _emittedToOffset=" + _emittedToOffset);
            failed.add(offset);
            numberFailed++;
            if (numberAcked == 0 && numberFailed > _spoutConfig.maxOffsetBehind) {
                throw new RuntimeException("Too many tuple failures");
            }
        }
    }

之前storm-kafka-0.8plus的版本号是这种（摘自storm-kafka-0.8-plus 源代码解析）

首先作者没有cache message。而仅仅是cache offset 
所以fail的时候，他是无法直接replay的，在他的凝视里面写了，不这样做的原因是怕内存爆掉

所以他的做法是，当一个offset fail的时候， 直接将_emittedToOffset回滚到当前fail的这个offset 
下次从Kafka fetch的时候会从_emittedToOffset開始读，这样做的优点就是依赖kafka做replay，问题就是会有反复问题 
所以使用时，一定要考虑，能否够接受反复问题

public void fail(Long offset) {
        //TODO: should it use in-memory ack set to skip anything that's been acked but not committed?
?
?

        // things might get crazy with lots of timeouts
        if (_emittedToOffset > offset) {
            _emittedToOffset = offset;
            _pending.tailSet(offset).clear();
        }
    }

大家对照一下吧

兴许继续更新上述未说清楚的地方，望大家包涵

reference

http://www.cnblogs.com/fxjwind/p/3808346.html