def task_done(self, message):
topic_partition = (message.topic, message.partition)<EOL>if topic_partition not in self._topics:<EOL><INDENT>logger.warning('<STR_LIT>'<EOL>'<STR_LIT>'.format(*topic_partition))<EOL>return False<EOL><DEDENT>offset = message.offset<EOL>prev_done = self._offsets.task_done[topic_partition]<EOL>if prev_done is not None and offset != (prev_done + <NUM_LIT:1>):<EOL><INDENT>logger.warning('<STR_LIT>',<EOL>offset, prev_done)<EOL><DEDENT>prev_commit = self._offsets.commit[topic_partition]<EOL>if prev_commit is not None and ((offset + <NUM_LIT:1>) <= prev_commit):<EOL><INDENT>logger.warning('<STR_LIT>',<EOL>offset, prev_commit)<EOL><DEDENT>self._offsets.task_done[topic_partition] = offset<EOL>if self._does_auto_commit_messages():<EOL><INDENT>self._incr_auto_commit_message_count()<EOL><DEDENT>if self._should_auto_commit():<EOL><INDENT>self.commit()<EOL><DEDENT>return True<EOL>
Mark a fetched message as consumed.

        Offsets for messages marked as "task_done" will be stored back
        to the kafka cluster for this consumer group on commit()

        Arguments:
            message (KafkaMessage): the message to mark as complete

        Returns:
            True, unless the topic-partition for this message has not
            been configured for the consumer. In normal operation, this
            should not happen. But see github issue 364.