KAFKA-5697: Use nonblocking poll in Streams #5107

vvcephei · 2018-05-31T15:57:12Z

Make use of the new Consumer#poll(Duration) to avoid getting stuck in poll when the broker is unavailable.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

vvcephei · 2018-05-31T15:58:27Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

@@ -81,7 +82,7 @@ public void register(final StateRestorer restorer) {

        final Set<TopicPartition> restoringPartitions = new HashSet<>(needsRestoring.keySet());
        try {
-            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(10);
+            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(Duration.ofMillis(10));


10ms isn't enough time to do any real work... It seems like it should either be 0ms (async) or >= 100ms like the rest.

I'd suggest using 100ms not 10ms, since this call is in a larger while loop in StreamThread.runOnce, and hence I concern doing poll(0) would unnecessarily increase the restoration latency.

Shouldn't we use config poll.ms? Or, if present, restore.consumer.poll.ms ?

Yeah, I was wondering why we sometimes use hardcoded values and other times use configured ones...

vvcephei · 2018-05-31T16:30:29Z

@bbejeck @guozhangwang @mjsax

Please review this when you get a chance.

bbejeck

Overall looks good to me. I just have a minor comment about using configuration for the timeouts in some spots.
Also left a comment about running system tests.

bbejeck · 2018-05-31T17:19:37Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -198,7 +199,7 @@ public void register(final StateStore store,
        int attempts = 0;
        while (true) {
            try {
-                partitionInfos = globalConsumer.partitionsFor(sourceTopic);
+                partitionInfos = globalConsumer.partitionsFor(sourceTopic, Duration.ofSeconds(30));


Do we want to consider using a configuration for this and other spots where we have hard-coded values? With KIP-276 merged users can specify different timeouts per consumer type.

I agree in general, so I'll make that change in the calls to poll, but I don't think we have an appropriate config for this call to partitionsFor. I just looked up the old call, and it does timeout according to a config. It's just that it waits 305s instead of 30s.

I think I'll revert this line and update the test to set that config lower instead.

bbejeck · 2018-05-31T17:20:05Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -262,7 +263,7 @@ private void restoreState(final StateRestoreCallback stateRestoreCallback,

            while (offset < highWatermark) {
                try {
-                    final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(100);
+                    final ConsumerRecords<byte[], byte[]> records = globalConsumer.poll(Duration.ofMillis(100));


Same as above and others as well.

bbejeck · 2018-05-31T17:22:18Z

streams/src/test/java/org/apache/kafka/streams/perf/SimpleBenchmark.java

@@ -334,7 +335,7 @@ private void consumeAndProduce(final String topic) {
            consumer.seekToBeginning(partitions);

            while (true) {
-                final ConsumerRecords<Integer, byte[]> records = consumer.poll(POLL_MS);
+                final ConsumerRecords<Integer, byte[]> records = consumer.poll(Duration.ofMillis(POLL_MS));


Just for sanity checking, maybe we should run SimpleBenchmark system tests?

Yeah, I definitely think we should run all the tests we can before the 2.0 release.

bbejeck · 2018-05-31T17:23:37Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

@@ -254,7 +255,7 @@ private static void ensureStreamsApplicationDown(final String kafka) {
                topics.add("repartition");
            }
            consumer.subscribe(topics);
-            consumer.poll(0);
+            consumer.poll(Duration.ofMillis(100));


the same comment as SimpleBenchmark, but thinking now maybe just run all system tests though.

This call is only to make sure we have updated the metadata. So I'm wondering if we can just get rid of this call, and instead on line 262 partitionFor now use request timeout as the duration, which should be sufficient?

Oh, I overlooked this. We should never call poll and ignore the result. I'll look into @guozhangwang's suggestion.

I'm just removing this poll call. As I understand it, the partitionsFor call (and all other interactions with Consumer) should be sufficient on their own. If they start timing out, we should just increase the timeout, IMO.

guozhangwang

Should we remove the two @ignore tags on KafkaStreamsTest class? cc @vvcephei

guozhangwang · 2018-05-31T18:21:50Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

@@ -81,7 +82,7 @@ public void register(final StateRestorer restorer) {

        final Set<TopicPartition> restoringPartitions = new HashSet<>(needsRestoring.keySet());
        try {
-            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(10);
+            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(Duration.ofMillis(10));


I'd suggest using 100ms not 10ms, since this call is in a larger while loop in StreamThread.runOnce, and hence I concern doing poll(0) would unnecessarily increase the restoration latency.

guozhangwang · 2018-05-31T18:26:23Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -1078,7 +1079,7 @@ private void maybeUpdateStandbyTasks(final long now) {
            }

            try {
-                final ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(0);
+                final ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(Duration.ofMillis(100));


Here I'd suggest doing the opposite: poll(0) since it is during the normal processing, not during restoration; so we can afford to not having some time in a few iterations. Instead, we want to proceed to the next iteration to call the normal-consumer.poll sooner to not be kicked out of the group.

guozhangwang · 2018-05-31T18:29:46Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

@@ -254,7 +255,7 @@ private static void ensureStreamsApplicationDown(final String kafka) {
                topics.add("repartition");
            }
            consumer.subscribe(topics);
-            consumer.poll(0);
+            consumer.poll(Duration.ofMillis(100));


This call is only to make sure we have updated the metadata. So I'm wondering if we can just get rid of this call, and instead on line 262 partitionFor now use request timeout as the duration, which should be sufficient?

mjsax

Thanks for the PR!

mjsax · 2018-06-01T05:23:07Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StoreChangelogReader.java

@@ -81,7 +82,7 @@ public void register(final StateRestorer restorer) {

        final Set<TopicPartition> restoringPartitions = new HashSet<>(needsRestoring.keySet());
        try {
-            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(10);
+            final ConsumerRecords<byte[], byte[]> allRecords = restoreConsumer.poll(Duration.ofMillis(10));


Shouldn't we use config poll.ms? Or, if present, restore.consumer.poll.ms ?

mjsax · 2018-06-01T05:27:59Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

-        streams.close();
+        try {
+            streams.start();
+        } catch (final StreamsException expected) {


If we expect this exception to happen, we should so a fail after stream.start()?

If we are afraid it might not timeout, but hang, should we put a timeout annotation to the test?

Ah, I was unaware of that annotation. I thought they all just timeout and fail after 30s.

I didn't add the fail because it's not important to verify that we get the exception. We are only verifying that the call doesn't hang.

I do think that we'll get the exception every time, so I could add the fail, I just chose to test the behavior we want (non-hanging), not the tightest bound on the behavior we observe (the exception). In other words, we could make a change in the future that doesn't throw the exception but instead exits normally, and that would be fine with this test. WDYT?

I think we should test for the exception -- if we change the behavior intentionally, we should remove the fail as well as the try-catch instead of allowing both behavior to pass (it's an either or form my point of view). IMHO, tests should be on an as narrow code path as possible: if we change behavior and a test fails, we are forces to reflect on the change, what is good as it guards against undesired behavior changes...

guozhangwang · 2018-06-04T20:25:18Z

The code change LGTM. @vvcephei have you tried to run the system tests? If yes could you paste its link here? If the system tests passed consistently and Jenkins works fine we can merge then.

vvcephei · 2018-06-04T20:27:40Z

@guozhangwang I'm doing it now. I'll let you know when it's done.

mjsax

Overall looks good. Couple of comments. Maybe @guozhangwang and @bbejeck want to chime in, too.

mjsax · 2018-06-05T20:07:14Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java

@@ -76,6 +78,7 @@ public GlobalStateManagerImpl(final LogContext logContext,
        this.stateRestoreListener = stateRestoreListener;
        this.retries = config.getInt(StreamsConfig.RETRIES_CONFIG);
        this.retryBackoffMs = config.getLong(StreamsConfig.RETRY_BACKOFF_MS_CONFIG);
+        this.pollTime = Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG));


This will return the value for "poll.ms" -- Should we check for "global.consumer.poll.ms" and user this value if present instead?

Hmm.. poll.ms is not a ConsumerConfig, but a StreamsConfig, so users are not expected to prefix it with the consumer.

I see what you mean. Still I am wondering if a single config for all consumers is smart. Also, the config is only used to pass into Consumer#poll() -- so even if it's not a config parameter, it's effectively still very similar to a "consumer config". I agree that it would required explicit documentation.

I agree it is generally better to allow different values for this; on the other hand, I felt it is too much of a lower-level details for people to really figure it out right. For example, today global consumers have two callers of poll(), restore consumer has two as well, and normal consumer has one. We can argue further that all these five calls may prefer different values for the reason we discussed in this PR, like whether it is call in the main loop or not.

Besides, if want to allow prefix for this config we can rush it into 2.0, but it may be considered a public API change as well.

Ack. I just convinced myself that it makes sense what you say. As you pointed out, we call poll() on different consumers with different purpose:

normale processing:

main-consumer

global-consumer (regular g-table update)

updating standby tasks:

restore-consumer

restoring:

main-consumer (that is still doing regular work)

restore-consumer

global-consumer

Applying the same value poll.ms to main and global consumer in normal processing case makes sense to me. For updating standby-tasks, we hard-code "zero" already. Same for the main-consumer during restore phase, to speed up restoring. Both are reasonable choices. The last case, is the restore case for restore-consumer and global-consumer -- it might make sense to apply a different poll-time for thiscase, but I also think that using the same as for the normal processing case is fine.

Thus, long story short, iff we ever introduce a prefix for poll.ms parameters, it should be for "normal processing phase" and "restore phase" but should not be tied to the actual used consumer-instance. Thus, using the consumer prefix does not make sense.

mjsax · 2018-06-05T20:09:33Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -212,7 +213,7 @@ State setState(final State newState) {
            if (newState == State.RUNNING) {
                updateThreadMetadata(taskManager.activeTasks(), taskManager.standbyTasks());
            } else {
-                updateThreadMetadata(Collections.<TaskId, StreamTask>emptyMap(), Collections.<TaskId, StandbyTask>emptyMap());
+                updateThreadMetadata(Collections.emptyMap(), Collections.emptyMap());


Thank you, Java8! :)

mjsax · 2018-06-05T20:10:16Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -602,7 +603,8 @@ public static StreamThread create(final InternalTopologyBuilder builder,
        log.info("Creating restore consumer client");
        final Map<String, Object> restoreConsumerConfigs = config.getRestoreConsumerConfigs(threadClientId);
        final Consumer<byte[], byte[]> restoreConsumer = clientSupplier.getRestoreConsumer(restoreConsumerConfigs);
-        final StoreChangelogReader changelogReader = new StoreChangelogReader(restoreConsumer, userStateRestoreListener, logContext);
+        final Duration pollTime = Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG));


As above: should we check for "restore.consumer.poll.ms" ?

I guess not, following Guozhang's comment.

mjsax · 2018-06-05T20:13:48Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -710,10 +712,10 @@ public StreamThread(final Time time,
        this.originalReset = originalReset;
        this.versionProbingFlag = versionProbingFlag;

-        this.pollTimeMs = config.getLong(StreamsConfig.POLL_MS_CONFIG);
+        this.pollTime = Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG));


as above: "main.consumer.poll.ms" ?

I guess not, following Guozhang's comment.

mjsax · 2018-06-05T20:15:17Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

-                final ConsumerRecords<byte[], byte[]> records = restoreConsumer.poll(0);
+                // poll(0): Since this is during the normal processing, not during restoration.
+                // We can afford to have slower restore (because we don't wait inside poll for results).
+                // Instead, we want to proceed to the next iteration to call the normal-consumer.poll


nit: the normal-consumer.poll -> main consumer#poll()

mjsax · 2018-06-05T20:15:34Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

+                // poll(0): Since this is during the normal processing, not during restoration.
+                // We can afford to have slower restore (because we don't wait inside poll for results).
+                // Instead, we want to proceed to the next iteration to call the normal-consumer.poll
+                // sooner to not be kicked out of the group.


nit: sooner -> as soon as possible

mjsax · 2018-06-05T20:17:03Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -234,7 +235,6 @@ public boolean conditionMet() {
        assertEquals(streams.state(), KafkaStreams.State.NOT_RUNNING);
    }

-    @Ignore // this test cannot pass as long as GST blocks KS.start()
    @Test
    public void testGlobalThreadCloseWithoutConnectingToBroker() {


rename: globalThreadShouldTimeoutAndThrowExceptionWhenBrokerConnectionCannotBeEstablished

mjsax · 2018-06-05T20:17:23Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

@@ -243,17 +243,22 @@ public void testGlobalThreadCloseWithoutConnectingToBroker() {
        props.setProperty(StreamsConfig.METRIC_REPORTER_CLASSES_CONFIG, MockMetricsReporter.class.getName());
        props.setProperty(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getPath());
        props.put(StreamsConfig.NUM_STREAM_THREADS_CONFIG, NUM_THREADS);
+        props.put("session.timeout.ms", 3001);
+        props.put(CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG, 3002);


why 3001 and 3002 ?

There are dependencies among the timing configs, session timeout must be greater than some other timeouts. But this test doesn't need to take 3s.

I'm pushing a change to be clearer about what's going on here and also reduce the time to 200ms.

mjsax · 2018-06-05T20:19:10Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

@@ -254,7 +256,7 @@ private static void ensureStreamsApplicationDown(final String kafka) {
                topics.add("repartition");
            }
            consumer.subscribe(topics);
-            consumer.poll(0);
+            firstPoll = consumer.poll(Duration.ofMillis(100));


vvcephei · 2018-06-06T14:48:13Z

Thanks for the review, @mjsax.
I need to wait until #5142 is merged, then I'll rebase this (and resolve the comment), and then we should be done with this PR.

mjsax · 2018-06-06T18:41:11Z

Still want to discuss the prefix thing. Also, this PR needs to be rebase because of #5013

guozhangwang · 2018-06-06T21:14:06Z

I'm +1 assuming that the PR is rebased.

vvcephei · 2018-06-06T21:38:48Z

Note to self: This will need a cherry-pick PR to 2.0 branch.

vvcephei · 2018-06-07T21:26:18Z

Fixed and rebased. The fix I needed was to switch to AdminClient to fetch the committed offsets in EosTestDriver. It would be worth re-reviewing that part.

New round of system tests in progress. I'll paste a link if/when they pass.

Also, refreshing the reminder to myself to cherry-pick this once it's merged.

guozhangwang

LGTM, assuming system test passed.

mjsax

Just some nits. Not reason to re-run the system tests.

For my own education: why did the EosTestDriver break? Unclear to me atm why using the consumer does not work any longer to get the committed offsets?

mjsax · 2018-06-08T01:37:19Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -843,15 +845,15 @@ long runOnce(final long recordsProcessedBeforeCommit) {
    /**
     * Get the next batch of records by polling.
     *
-     * @param pollTimeMs poll time millis parameter for the consumer poll
+     * @param pollTime poll time millis parameter for the consumer poll


nit: update parameter description; still says "millis"

mjsax · 2018-06-08T01:38:35Z

streams/src/test/java/org/apache/kafka/streams/KafkaStreamsTest.java

+        // few other configs.
+        props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 200);
+        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 200);
+        props.put("session.timeout.ms", 201);


nit: use ConsumerConfig .SESSION_TIMEOUT_MS_CONFIG

oh, weird, I was looking for that config, and the only one I found was in an internal class.

mjsax · 2018-06-08T01:41:02Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

-                for (final PartitionInfo partition : consumer.partitionsFor(topic)) {
-                    partitions.add(new TopicPartition(partition.topic(), partition.partition()));
-                }
+        try (final org.apache.kafka.clients.admin.AdminClient adminClient = KafkaAdminClient.create(props)) {


nit: use class import to avoid long name here

The scala AdminClient was already in scope. I've refactored to just use the Java one.

mjsax · 2018-06-08T01:41:55Z

streams/src/test/java/org/apache/kafka/streams/tests/EosTestDriver.java

+            final ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult = adminClient.listConsumerGroupOffsets(EosTestClient.APP_ID);
+            try {
+                topicPartitionOffsetAndMetadataMap = listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().get();
+            } catch (final InterruptedException | ExecutionException e) {


Why this nested try-catch? Can't we use a single one?

vvcephei · 2018-06-08T14:10:23Z

Wow, the system tests passed!
results: http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2018-06-07--001.1528440548--vvcephei--KAFKA-5697-streams-nonblocking-poll--966aa7b/report.html
build: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1775

The control run on current trunk had some issues:
results: http://confluent-kafka-system-test-results.s3-us-west-2.amazonaws.com/2018-06-07--001.1528441472--apache--trunk--287a92f/report.html
build: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1776

Were there any other tests you wanted to see?

vvcephei · 2018-06-08T15:11:31Z

With this latest change, I'm going to run the EOS system tests again.

vvcephei · 2018-06-08T15:48:37Z

The EOS system tests passed for me locally. Here's the in-progress run of the EOS tests x5: https://jenkins.confluent.io/job/system-test-kafka-branch-builder/1780

vvcephei · 2018-06-08T16:29:47Z

In response to your question, @mjsax,

The call to Consumer#position requires that the consumer instance has already joined the consumer group. The only call in Consumer that will cause it to join the group is poll. Therefore, calling poll is a precondition to calling position. We used to call poll(0), but that's deprecated now, for good reasons. We cannot call poll(Duration) because the intent right here is just to find the committed offsets, specifically not to poll any messages.

We have considered adding a separate method to Consumer to joinGroupIfNeeded or something, and I actually could have created it protected to support internal usage of it. Another option was to make position join the group directly.

Upon reflection, though, both of those are silly for this usage. Why should we create a new Consumer instance and join the group when we intend not to poll; only to find out what the committed offsets are? That was the only operation we were doing on the client before closing it. This operation is more appropriately done via the AdminClient, which is what we're doing now.

And this is ultimately why we opted not to add that extra Consumer method right away. Although it wouldn't be wrong to have it, it's not clear that it's needed. At first (and now second) glance, the only reason you should join the group is to poll messages, so we avoid misleadingly implying otherwise by omitting the (presumably useless) method.

vvcephei · 2018-06-08T17:08:31Z

Ok, the EOS system tests still pass: http://confluent-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/2018-06-08--001.1528477413--vvcephei--KAFKA-5697-streams-nonblocking-poll--b829d55/report.html

Call for final reviews @bbejeck @mjsax @guozhangwang .

Make use of the new Consumer#poll(Duration) to avoid getting stuck in poll when the broker is unavailable. Reviewers: Matthias J. Sax <[email protected]>, Guozhang Wang <[email protected]>, Bill Bejeck <[email protected]>

guozhangwang

Cherry-picked to 2.0 since it is an important bug fix to improve operational experience.

vvcephei · 2018-06-10T00:20:45Z

Thanks, @guozhangwang !

Make use of the new Consumer#poll(Duration) to avoid getting stuck in poll when the broker is unavailable. Reviewers: Matthias J. Sax <[email protected]>, Guozhang Wang <[email protected]>, Bill Bejeck <[email protected]>

vvcephei commented May 31, 2018

View reviewed changes

bbejeck reviewed May 31, 2018

View reviewed changes

guozhangwang reviewed May 31, 2018

View reviewed changes

mjsax reviewed Jun 1, 2018

View reviewed changes

mjsax reviewed Jun 5, 2018

View reviewed changes

vvcephei mentioned this pull request Jun 5, 2018

KAFKA-7000: update assignment in Consumer#position #5142

Closed

3 tasks

guozhangwang reviewed Jun 8, 2018

View reviewed changes

mjsax reviewed Jun 8, 2018

View reviewed changes

KAFKA-5697: Use nonblocking poll in Streams

b829d55

guozhangwang merged commit 74bdafe into apache:trunk Jun 8, 2018

guozhangwang reviewed Jun 8, 2018

View reviewed changes

vvcephei deleted the KAFKA-5697-streams-nonblocking-poll branch June 10, 2018 00:20

KAFKA-5697: Use nonblocking poll in Streams #5107

KAFKA-5697: Use nonblocking poll in Streams #5107

Conversation

vvcephei commented May 31, 2018

Committer Checklist (excluded from commit message)

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

vvcephei commented May 31, 2018

bbejeck 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

Choose a reason for hiding this comment

guozhangwang left a comment • 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

mjsax left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei Jun 1, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

guozhangwang commented Jun 4, 2018

vvcephei commented Jun 4, 2018

mjsax 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

mjsax Jun 5, 2018 • 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vvcephei commented Jun 6, 2018

mjsax commented Jun 6, 2018

guozhangwang commented Jun 6, 2018

vvcephei commented Jun 6, 2018

vvcephei commented Jun 7, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

mjsax 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

vvcephei commented Jun 8, 2018

vvcephei commented Jun 8, 2018

vvcephei commented Jun 8, 2018 • edited Loading

vvcephei commented Jun 8, 2018 • edited Loading

vvcephei commented Jun 8, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

vvcephei commented Jun 10, 2018

guozhangwang left a comment •

edited

Loading

vvcephei Jun 1, 2018 •

edited

Loading

mjsax Jun 5, 2018 •

edited

Loading

vvcephei commented Jun 8, 2018 •

edited

Loading

vvcephei commented Jun 8, 2018 •

edited

Loading