KAFKA-3853 (KIP-88): Report offsets for empty groups in consumer group describe command (new consumer) #2074

vahidhashemian · 2016-10-28T20:37:09Z

When there are no active new consumers in a consumer group report the offsets within the group instead of reporting that the group has no active members.

This PR also implements the API change proposed by KIP-88.

vahidhashemian · 2016-10-28T20:42:10Z

@hachikuji This PR also fixes another issue that was introduced by the fix for KAFKA-3144. I point to it inline. Let me know if you think that issue should be fixed and merged (more quickly) in a separate PR. Thanks.

vahidhashemian · 2016-10-28T20:44:23Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

              case Some(other) =>
                // the control should never reach here
                throw new KafkaException(s"Expected a valid consumer group state, but found '$other'.")
-              case None =>
-                // the control should never reach here
-                throw new KafkaException("Expected a valid consumer group state, but none found.")


This is where the bug was introduced. If state is None, there is a possibility that the old consumer based group does not have any active members; so we need to check whether new consumer is used or not, and then proceed accordingly.

hachikuji · 2016-10-28T20:44:33Z

clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java

    public static final Schema[] DESCRIBE_GROUPS_REQUEST = new Schema[] {DESCRIBE_GROUPS_REQUEST_V0};
-    public static final Schema[] DESCRIBE_GROUPS_RESPONSE = new Schema[] {DESCRIBE_GROUPS_RESPONSE_V0};
+    public static final Schema[] DESCRIBE_GROUPS_RESPONSE = new Schema[] {DESCRIBE_GROUPS_RESPONSE_V1};


Was there a KIP for this that I missed?

Thanks for bringing this up. I wasn't totally sure what to process is for changing the protocol, and whether I'm actually correct to assume that the protocol has to be changed for this JIRA. I haven't opened a KIP yet, if you think that's eventually going to be needed for I'd be happy to create one. Thanks in advance for clarifying this.

Yeah, protocol changes definitely need a KIP. Probably makes sense then to split the bug fix into a separate patch.

Thanks. I'll submit the bug fix separately, and then work on creating a KIP.

vahidhashemian · 2016-11-04T23:11:13Z

core/src/test/scala/unit/kafka/admin/DescribeConsumerGroupTest.scala

+    props.setProperty("key.deserializer", classOf[ByteArrayDeserializer].getName)
+    props.setProperty("value.deserializer", classOf[ByteArrayDeserializer].getName)
+    val consumer1Mock = EasyMock.createMockBuilder(classOf[NewShinyConsumer]).withConstructor(Some(topic), None, None, None, props, Long.MaxValue.asInstanceOf[Object]).createMock()
+    val consumer2Mock = EasyMock.createMockBuilder(classOf[NewShinyConsumer]).withConstructor(Some(topic), None, None, None, props, Long.MaxValue.asInstanceOf[Object]).createMock()


@hachikuji I was hoping you could take a look at an issue I'm running into if and when you get a chance.
While KIP-88 is open for discussion I spent some time creating some unit tests for this PR.

This particular unit test simply mocks two consumers that consume from the same 1-partition topic and belong to the same consumer group. A similar unit test on old consumers exists earlier in the same file and runs fine. There are also other unit tests above using the new consumer that run fine but they mock only one consumer.

The problem I'm running into is this line that mocks the second consumer and takes a long time to run (I believe for the consumer to join the group that eventually times out) and the consumer group gets corrupted somehow. When I debug and check the status of the group down in the waitUntilTrue check, sometimes it is Empty, or Dead, or even Stable with only one of the consumers and it never gets into the expected state (Stable with two members). Where it gets stuck in a loop I think is here (after trying to join the group).

I'm not sure if I'm doing something wrong with the unit test or if I'm hitting some bug. I thought you might know by just looking at it. Thanks.

asfbot · 2016-12-12T18:50:01Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/83/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2016-12-12T21:45:28Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/81/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2016-12-12T21:45:37Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/82/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2016-12-13T20:37:12Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/112/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2016-12-13T20:37:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/111/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2016-12-13T20:39:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/110/
Test FAILed (JDK 7 and Scala 2.10).

vahidhashemian · 2016-12-13T20:41:13Z

@hachikuji I've tried to address both KIP-88 and KAFKA-3853 in the same PR here. I'd appreciate your feedback when you get a chance to review this. Thanks.

asfbot · 2017-01-13T04:01:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/827/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T04:09:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/825/
Test PASSed (JDK 7 and Scala 2.10).

hachikuji · 2017-01-13T04:57:23Z

@vahidhashemian Saw your question about why we are only checking 3 of the 5 error codes. Haha, I noticed the same thing when reviewing and opened KAFKA-4622.

vahidhashemian · 2017-01-13T05:10:48Z

Thanks for clarifying. Then I can leave that to be fixed as part of that JIRA.

hachikuji

This is getting close, but I think there's still a problem with the handler logic in KafkaApis.

hachikuji · 2017-01-13T05:09:43Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

@@ -47,6 +48,14 @@ public Builder(String groupId, List<TopicPartition> partitions) {
            this.partitions = partitions;
        }

+        public static Builder allTopicPartitions(String groupId) {


Maybe we should enforce a minimum version number when querying all partitions? You can look at ListOffsetRequest for an example of this.

hachikuji · 2017-01-13T05:16:00Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

+        short versionId = version();
+
+        if (versionId < 2 && partitions != null) {
+            for (TopicPartition partition: partitions) {


nit: add a space before the :.

hachikuji · 2017-01-13T05:20:11Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java

+            this.error = topLevelError;
+            struct.set(ERROR_CODE_KEY_NAME, this.error.code());
+        } else
+            this.error = Errors.NONE;


Wonder if there's any harm retaining the top-level error regardless of the version. Seems more consistent with how we handle the case of constructing from a Struct.

I think it should be okay to do that. Will update.

hachikuji · 2017-01-13T05:23:09Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java

                responseData.put(new TopicPartition(topic, partition), partitionData);
            }
        }
+
+        this.error = struct.hasField(ERROR_CODE_KEY_NAME) ? Errors.forCode(struct.getShort(ERROR_CODE_KEY_NAME)) : topLevelError;


Probably worth a comment explaining why we do this.

hachikuji · 2017-01-13T05:25:39Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

@@ -66,60 +75,81 @@ public String toString() {
    private final String groupId;
    private final List<TopicPartition> partitions;

-    // v0 and v1 have the same fields.
+    public static OffsetFetchRequest forAllPartitions(String groupId, short version) {


Related to above comment. This method only makes sense for version 2, so maybe we should remove version and use 2 directly.

hachikuji · 2017-01-13T05:36:09Z

core/src/main/scala/kafka/coordinator/GroupCoordinator.scala

-      partitions.map { topicPartition =>
-        (topicPartition, new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, "", Errors.GROUP_COORDINATOR_NOT_AVAILABLE.code))}.toMap
-    } else if (!isCoordinatorForGroup(groupId)) {
+                         partitions: Seq[TopicPartition]): (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) = {


Instead of using null as the sentinel, we could use an Option.

hachikuji · 2017-01-13T05:37:41Z

core/src/main/scala/kafka/coordinator/GroupCoordinator.scala

      // return offsets blindly regardless the current group state since the group may be using
      // Kafka commit storage without automatic group management
-      groupManager.getOffsets(groupId, partitions)
+      (Errors.NONE, groupManager.getOffsets(groupId, partitions match {


Another option would be to push the handling of all offsets into getOffsets. One small advantage is that you would only need to acquire the lock once instead of twice.

Great idea because getOffsets already gives us what we want (here).

hachikuji · 2017-01-13T06:11:25Z

core/src/main/scala/kafka/server/KafkaApis.scala

+        }
+      } else {
+        val isAllPartitions = offsetFetchRequest.isAllPartitions() // header.apiVersion == 2
+        var groupOffsets: (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) = null


You could do these assignments at once:

val (groupOffsets, partitions) = ...

We use this pattern just below.

Would it also work if one assignment depends on the other one? partitions uses groupOffsets. Also, it may not read easily since there is a big type definition for groupOffsets.

Does type inference not work? I was thinking something like this:

val (groupOffsets, partitions) = if (isAllPartitions) { val groupOffsets = coordinator.handleFetchOffsets(offsetFetchRequest.groupId, null) (groupOffsets, groupOffsets._2.keySet) } else { val requestPartitions = offsetFetchRequest.partitions.asScala.toList (coordinator.handleFetchOffsets(offsetFetchRequest.groupId, requestPartitions), requestPartitions) }

Yes, it works perfectly. Sorry for the premature question!

hachikuji · 2017-01-13T06:13:31Z

core/src/main/scala/kafka/server/KafkaApis.scala

+        var groupOffsets: (Errors, Map[TopicPartition, OffsetFetchResponse.PartitionData]) = null
+        val partitions =
+          if (isAllPartitions) {
+            groupOffsets = coordinator.handleFetchOffsets(offsetFetchRequest.groupId, null)


Seems like this and the other call to handleFetchOffsets below needs to go to the else case after the check for version 0. For version 0, we pull offsets from zookeeper. I'm wondering if your first approach, which collected all partitions from the coordinator first, may have been a little cleaner. Another option to consider, perhaps you could do the post-filtering for the isAllPartitions case separately, and continue doing pre-filtering when we are provided the partition list.

hachikuji · 2017-01-13T06:17:07Z

core/src/main/scala/kafka/server/KafkaApis.scala

-                  (topicPartition, unknownTopicPartitionResponse)
+      // reject the request if not authorized to the group
+      if (!authorize(request.session, Read, new Resource(Group, offsetFetchRequest.groupId))) {
+        if (header.apiVersion >= 2)


I was thinking we could handle both of these cases in the same constructor. The constructor would take a single error code and the list of requested partitions. If the version is greater than or equal to 2, the partitions are ignored; otherwise, the error code is written into the partition errors.

Good idea. Thanks.

hachikuji · 2017-01-13T18:15:21Z

@vahidhashemian I merged KIP-103, which looks like it causes some conflicts in ApiVersion. Hopefully shouldn't be too much trouble to resolve.

hachikuji · 2017-01-13T19:42:08Z

core/src/main/scala/kafka/server/KafkaApis.scala

+          new OffsetFetchResponse(results.asJava, Errors.NONE, header.apiVersion)
+        }
+      } else {
+        val isAllPartitions = offsetFetchRequest.isAllPartitions() // header.apiVersion == 2


nit: unneeded parenthesis.

vahidhashemian · 2017-01-13T22:21:35Z

clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java

-                        coordinatorDead();
-                        future.raise(error);
-                    } else if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
+                    if (error == Errors.UNKNOWN_TOPIC_OR_PARTITION) {
                        future.raise(new KafkaException("Partition " + tp + " may not exist or user may not have Describe access to topic"));


Still not too sure about this. Did you want to move this error check up in the first if block?

I'm not sure I see the problem. Would we ever see this at the top level?

I'm referring to your earlier comment, and I'm not sure if it implied modifying this method too.

I was mainly concerned that we'd need to check errors in both places, but I think we're good now since we ensure that top-level error codes will always appear at the top level (even for older versions).

Great, thanks for clarifying.

hachikuji · 2017-01-13T22:46:58Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchRequest.java

        @Override
        public OffsetFetchRequest build() {
+            if (isAllTopicPartitions() && version() < minVersion)


Actually I guess we could make this a little simpler. We know we need version 2, so maybe we can use it directly and remove minVersion?

Sure, I thought about it too, but thought to keep it in sync with ListOffsetRequest. I'll update.

Yeah, we're still feeling out the best patterns for handling older versions.

hachikuji · 2017-01-13T23:04:02Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java

+
+        // for version 2 and later use the top level error code (in ERROR_CODE_KEY_NAME) from the response
+        // for older versions there is no top level error in the response and all errors are partition errors,
+        // so if there is a group or coordinator error at the partition level use that as the top level error


Could we mention that we do this so that the client can depend on the top-level error code regardless of the offset fetch version?

asfbot · 2017-01-13T23:05:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/866/
Test PASSed (JDK 8 and Scala 2.11).

hachikuji · 2017-01-13T23:05:30Z

clients/src/main/java/org/apache/kafka/common/requests/OffsetFetchResponse.java

    private final Map<TopicPartition, PartitionData> responseData;
+    private final Errors error;
+    private Set<Errors> partitionErrors;


Do we need another field if the errors are already contained in PartitionData?

I added this to keep track of partition errors that is needed by AdminClient here. Unless it's okay to process PartitionData on the fly?

Seems just as efficient to me, especially since we only throw the first error.

Sure, then I think we perhaps need to at least have another Errors member for that first partition error. So we don't have to process partitionData multiple times for checking the existence and actually retrieving the error (in hasPartitionErrors and partitionErrors).

Hmm.. It just doesn't seem worth optimizing for. Processing the partition data means what? Looping over it and checking if error is NONE? Does it matter if we do that twice? We could also just leave off the hasPartitionErrors and do a single iteration and raise the error on the first exception.

Would something like this work?

public Errors getPartitionError() { Collection<PartitionData> partitionsData = this.responseData.values(); for (PartitionData data : partitionsData) { if (data.hasError()) return data.error; } return null; }

Sure, that would work. Maybe getFirstPartitionError is a clearer name? Or you could bundle the exception throwing as well into a single maybeThrowFirstPartitionError? Either way is fine with me, but I'd prefer not to additional fields without a clear case that they're needed.

hachikuji · 2017-01-13T23:10:41Z

core/src/main/scala/kafka/api/ApiVersion.scala

@@ -146,3 +148,9 @@ case object KAFKA_0_10_2_IV0 extends ApiVersion {
  val messageFormatVersion: Byte = Record.MAGIC_VALUE_V1
  val id: Int = 9
 }
+
+case object KAFKA_0_10_2_IV1 extends ApiVersion {


Talked to @ijuma about this, and I don't think we need to bump the internal version number since the brokers do not use offset fetches themselves.

Sounds good. I'll remove the internal version.

hachikuji · 2017-01-13T23:28:38Z

core/src/main/scala/kafka/server/KafkaApis.scala

+                None) // we are not interested in all offsets in the group here
+          } else {
+            // the request is for all group offsets (api version 2 and later)
+            val groupOffsets = coordinator.handleFetchOffsets(offsetFetchRequest.groupId, None)


I think this is close, but it's a bit annoying that we have to call handleFetchOffsets in two places, right? Wouldn't it be better to delay the check for isAllPartitions and the filtering until after the call to handleFetchOffsets below. That gives us a clean separation of the Kafka and Zookeeper offset handling. So maybe the logic can be something like this:

Filter unauthorized partitions

Check if this is version 0
a. Fetch from zk for version 0
b. Check from kafka for versions 1 and above. After receiving the fetched offsets, check if isAllPartitions is set. If so, additionally filter out the fetched offsets for topics we are not authorized for.

Does that make sense?

If I understand this correctly, for version 1 and above, to receive the fetched offsets we already need to check isAllPartitions to determine if None or Some(partitions) should be passed to handleFetchOffsets.

Yeah, that's fair. The point is that it should happen in the Kafka branch of that if and not before.

I think we need to call isAllPartitions upfront anyway because we need to make sure offsetFetchRequest.partitions is not null before starting to filter.

…p describe command (new consumer) When there are no active new consumers in a consumer group report the offsets within the group instead of reporting that the group has no active members. This PR also implements the API change proposed by [KIP-88](https://cwiki.apache.org/confluence/display/KAFKA/KIP-88%3A+OffsetFetch+Protocol+Update).

vahidhashemian · 2017-01-14T00:28:49Z

core/src/main/scala/kafka/server/KafkaApis.scala

+          }
+          else
+            new OffsetFetchResponse(offsets._1)
+        }


I hope this is now closer to what you described.

hachikuji

LGTM. Thanks for being so responsive throughout the review iterations!

asfbot · 2017-01-14T01:15:49Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/875/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-14T01:17:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/864/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-14T01:18:39Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/864/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-14T03:28:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/873/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-14T03:28:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/873/
Test FAILed (JDK 8 and Scala 2.12).

…a consumer group (KIP-88) Author: Vahid Hashemian <vahidhashemian@us.ibm.com> Reviewers: Ismael Juma <ismael@juma.me.uk>, Ewen Cheslack-Postava <ewen@confluent.io>, Jason Gustafson <jason@confluent.io> Closes apache#2074 from vahidhashemian/KAFKA-3853

vahidhashemian commented Oct 28, 2016

View reviewed changes

hachikuji reviewed Oct 28, 2016

View reviewed changes

vahidhashemian changed the title ~~KAFKA-3853: Report offsets for empty groups in consumer group describe command (new consumer)~~ KAFKA-3853 (WIP): Report offsets for empty groups in consumer group describe command (new consumer) Oct 28, 2016

vahidhashemian mentioned this pull request Oct 31, 2016

KAFKA-4357: Fix consumer group describe output when there is no active member (old consumer) #2075

Closed

vahidhashemian force-pushed the KAFKA-3853 branch 3 times, most recently from 3d733d8 to 5b1c5fc Compare November 4, 2016 22:59

vahidhashemian commented Nov 4, 2016

View reviewed changes

vahidhashemian force-pushed the KAFKA-3853 branch 3 times, most recently from 0198627 to 74cae9f Compare November 15, 2016 21:26

vahidhashemian force-pushed the KAFKA-3853 branch 7 times, most recently from 5ef58b9 to 6a3e835 Compare November 30, 2016 03:30

vahidhashemian force-pushed the KAFKA-3853 branch from 6a3e835 to 84cb611 Compare December 2, 2016 21:39

vahidhashemian force-pushed the KAFKA-3853 branch from 84cb611 to 94e8bab Compare December 12, 2016 18:44

vahidhashemian changed the title ~~KAFKA-3853 (WIP): Report offsets for empty groups in consumer group describe command (new consumer)~~ KAFKA-3853 (KIP-88): Report offsets for empty groups in consumer group describe command (new consumer) Dec 12, 2016

vahidhashemian force-pushed the KAFKA-3853 branch from 94e8bab to 498e8df Compare December 13, 2016 19:55

hachikuji reviewed Jan 13, 2017

View reviewed changes

vahidhashemian force-pushed the KAFKA-3853 branch 2 times, most recently from a6a3220 to ad48ee9 Compare January 13, 2017 22:16

vahidhashemian commented Jan 13, 2017

View reviewed changes

hachikuji reviewed Jan 13, 2017

View reviewed changes

vahidhashemian force-pushed the KAFKA-3853 branch from ad48ee9 to 5579ef9 Compare January 14, 2017 00:27

vahidhashemian commented Jan 14, 2017

View reviewed changes

hachikuji approved these changes Jan 14, 2017

View reviewed changes

asfgit closed this in c2d9b95 Jan 14, 2017

vahidhashemian mentioned this pull request Jan 31, 2017

MINOR: Use an explicit Errors object when possible instead of a numeric error code #2475

Closed

gena01 mentioned this pull request Jul 26, 2017

Support 0.10.2.X versions yahoo/CMAK#394

Open

efeg pushed a commit to efeg/kafka that referenced this pull request May 29, 2024

Upgrade org.json to 20231013+ due to CVE-2023-5072 (apache#2074)

c5f4944

KAFKA-3853 (KIP-88): Report offsets for empty groups in consumer group describe command (new consumer) #2074

KAFKA-3853 (KIP-88): Report offsets for empty groups in consumer group describe command (new consumer) #2074

Conversation

vahidhashemian commented Oct 28, 2016 • edited Loading

vahidhashemian commented Oct 28, 2016

vahidhashemian Oct 28, 2016 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian Nov 4, 2016 • edited Loading

Choose a reason for hiding this comment

asfbot commented Dec 12, 2016

asfbot commented Dec 12, 2016

asfbot commented Dec 12, 2016

asfbot commented Dec 13, 2016

asfbot commented Dec 13, 2016

asfbot commented Dec 13, 2016

vahidhashemian commented Dec 13, 2016

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

hachikuji commented Jan 13, 2017

vahidhashemian commented Jan 13, 2017

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian Jan 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Jan 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji commented Jan 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vahidhashemian Jan 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Jan 13, 2017 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji left a comment

Choose a reason for hiding this comment

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

vahidhashemian commented Oct 28, 2016 •

edited

Loading

vahidhashemian Oct 28, 2016 •

edited

Loading

vahidhashemian Nov 4, 2016 •

edited

Loading

vahidhashemian Jan 13, 2017 •

edited

Loading

hachikuji Jan 13, 2017 •

edited

Loading

vahidhashemian Jan 13, 2017 •

edited

Loading

hachikuji Jan 13, 2017 •

edited

Loading