Kafka__consumer_offsets主题具有过多的分区计数

Question

我使用的是Kafka 0.8.2，而我的使用者却出现了一个错误：“OFFSET commit FAILL with...”。当查看主题“__consumer_offsets”时。我看到它有50个分区计数。正常吗？我只能通过删除所有的Kafka日志并重新启动我的Kafka服务器来解决这个问题。是否有一种方法，我可以删除这个主题时，它达到一定数量的分区，还是我提交的偏移量是错误的？

下面是我提交偏移的方式：

 public void commitOffsets(BlockingChannel channel, String topic, String    groupid, int partition, String clientName, int corrilationid, long offset)   throws Exception{

    if (commitTryCount > 100){
        throw new Exception("Offset commit failed with " + channel.host());
    }

    long now = System.currentTimeMillis();
    Map<TopicAndPartition, OffsetAndMetadata> offsets = new LinkedHashMap<TopicAndPartition, OffsetAndMetadata>();
    //for (int i = 0; i < this.totalPartitions; i++){
        TopicAndPartition topicPartition = new TopicAndPartition(topic, partition);
        offsets.put(topicPartition, new OffsetAndMetadata(offset, topic, now));
    //}     

    //initialize offset commit
    OffsetCommitRequest commitRequest = new OffsetCommitRequest(groupid, offsets, corrilationid, clientName, (short) 1);
    channel.send(commitRequest.underlying());
    OffsetCommitResponse commitResponse = OffsetCommitResponse.readFrom(channel.receive().buffer());
    if (commitResponse.hasError()){         
        for (Object partitionErrorCode: commitResponse.errors().values()){
            if (Short.parseShort(partitionErrorCode.toString()) == ErrorMapping.OffsetMetadataTooLargeCode()){
                //reduce the size of the metadata and retry
                offset--;
                commitOffsets(channel, topic, groupid, partition, clientName, corrilationid, offset);
                commitTryCount++;
            } else if (Short.parseShort(partitionErrorCode.toString()) == ErrorMapping.NotCoordinatorForConsumerCode()
                    || Short.parseShort(partitionErrorCode.toString()) == ErrorMapping.ConsumerCoordinatorNotAvailableCode()) {
                //discover new coordinator and retry
                int newCorrilation = corrilationid;
                newCorrilation++;
                this.channel = discoverChannel(channel.host(), port, groupid, clientName, newCorrilation);
                commitOffsets(this.channel, topic, groupid, partition, clientName, newCorrilation, offset);
                commitTryCount++;
            } else{
                //retry
                commitOffsets(channel, topic, groupid, partition, clientName, corrilationid, offset);
                commitTryCount++;
            }//end of else              
        }//end of for
    }//end of if
}//end of method

Answer 1

我把代码贴出来后才弄明白的。当提交成功时，我忘记将变量“committrycount”设置为0。我仍然想知道__consumer_offsets主题有50个分区是否正常？