Remove consumer.listTopics() method in case when too many topics in kafka causes the FullGC in Overlord by elloooooo · Pull Request #6455 · apache/druid

elloooooo · 2018-10-12T08:33:53Z

When there are too many topics in kafka cluster, the method consumer.listTopics() will leads to Full GC in Overlord.

clintropolis

This does seem like it would be more efficient than pulling the list of all topics 👍, but Travis failures look perhaps related to this PR, maybe something to do with the removal of the lock?

clintropolis · 2018-10-12T17:05:33Z

-    }
-
-    List<PartitionInfo> partitions = topics.get(ioConfig.getTopic());
+    List<PartitionInfo> partitions = consumer.partitionsFor(ioConfig.getTopic());


Why did the lock and exception handling get removed? The lock seems needed, and checking the docs partitionsFor can throw a handful of exceptions too:

Throws: WakeupException - if wakeup() is called before or while this function is called InterruptException - if the calling thread is interrupted before or while this function is called AuthorizationException - if not authorized to the specified topic TimeoutException - if the topic metadata could not be fetched before expiration of the configured request timeout KafkaException - for any other unrecoverable errors

Thank you for reviewing this. I aggree that the lock and exception handing should not be removed. I will fix later.

b-slim · 2018-10-13T23:39:18Z

Dumb question why do we even need to list all topics if the user supply the input topic? Can this operation be removed.

b-slim · 2018-10-13T23:44:16Z

@elloooooo am wondering how big the topic list you guys had to see this issue in practice thanks!

elloooooo · 2018-10-15T01:35:10Z

@b-slim Someone used to create 20000+ topics at one of our kafka clusters which supports creating topic when needed. Actually his code has bugs.

…nd remove some useless checks

jihoonson · 2018-10-15T02:03:32Z

-      log.warn("No such topic [%s] found, list of discovered topics [%s]", ioConfig.getTopic(), topics.keySet());
-    }
-    int numPartitions = (partitions != null ? partitions.size() : 0);
+    int numPartitions = partitions.size();


Looks like partitions can be null. Should be int numPartitions = (partitions != null ? partitions.size() : 0);.

I think the partitions can be null only when consumer.partitionsFor(ioConfig.getTopic()) throw exceptions. And the exception has been catched and the function get return. So when partitions is null, there is no chance run int numPartitions = partitions.size();, is it?

This is the code of KafkaConsumer.partitionsFor().

public List<PartitionInfo> partitionsFor(String topic) { acquire(); try { Cluster cluster = this.metadata.fetch(); List<PartitionInfo> parts = cluster.partitionsForTopic(topic); if (parts != null) return parts; Map<String, List<PartitionInfo>> topicMetadata = fetcher.getTopicMetadata( new MetadataRequest.Builder(Collections.singletonList(topic)), requestTimeoutMs); return topicMetadata.get(topic); } finally { release(); } }

Since topicMetadata is a map, topicMetadata.get(topic); can return null.

Ah~ I didn't notice that. Thank you. I will fix it.

jihoonson · 2018-10-15T05:46:44Z

@elloooooo thanks for the quick fix! The Travis fail looks legit. Would you please fix it?

[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.2:testCompile (default-testCompile) on project druid-sql: Compilation failure: Compilation failure: 
[ERROR] /home/travis/build/apache/incubator-druid/sql/src/test/java/org/apache/druid/sql/calcite/util/CalciteTests.java:[588,9] incompatible types: java.lang.String cannot be converted to org.apache.druid.discovery.NodeType
[ERROR] /home/travis/build/apache/incubator-druid/sql/src/test/java/org/apache/druid/sql/calcite/util/CalciteTests.java:[585,49] cannot find symbol
[ERROR]   symbol: constructor (org.apache.druid.java.util.http.client.HttpClient,org.apache.druid.discovery.DruidNodeDiscoveryProvider,java.lang.String,java.lang.String,org.apache.druid.curator.discovery.ServerDiscoverySelector)

jihoonson · 2018-10-15T06:48:32Z

@elloooooo NVM. Looks like it's caused by the bug which #6466 fixes.

jihoonson · 2018-10-15T10:50:01Z

Would you please check the CI failure?

Tests run: 89, Failures: 17, Errors: 8, Skipped: 0, Time elapsed: 57.429 sec <<< FAILURE! - in org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorTest
testResetNoTasks[numThreads = 1](org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorTest)  Time elapsed: 0.155 sec  <<< FAILURE!
java.lang.AssertionError: 
  Unexpected method call IndexerMetadataStorageCoordinator.getDataSourceMetadata("testDS"):

This reverts commit f839d09.

elloooooo · 2018-10-18T02:07:00Z

@jihoonson I can't figure it out why the travis-ci failed.
Actually, there is no error like this in my local test.

Running org.apache.druid.curator.announcement.AnnouncerTest
No output has been received in the last 10m0s, this potentially indicates a stalled build or something wrong with the build itself.

Could you give me some suggestions？

b-slim · 2018-10-18T02:27:40Z

can you try to run druid-server tests on your local machine? seems to be a locking issue, but not sure. i have restarted the build anyway.

gianm · 2018-10-18T23:07:35Z

Hmm, they got stuck again but I restarted them again.

jihoonson · 2018-10-19T06:02:07Z

  private void addSomeEvents(int numEventsPerPartition) throws Exception
  {
+    //create topic manually
+    AdminUtils.createTopic(zkUtils, topic, NUM_PARTITIONS, 1, new Properties(), RackAwareMode.Enforced$.MODULE$);


Hmm, would you tell me why this is needed?

When the value of setting auto.craete.topics.enable is ture(default), consumer.partitionsFor(SOME_TOPIC) will create the SOME_TOPIC if the topic doesn't exist. This leads some UT like testXXXNoTasks() failed. So I disable it but this lead some UTs which need sending some events to kafka fail. So I create the topic in kafka manually before sending data.

Haha, it looks like there was something wrong in Github. Got it. Thanks.

Sorry about that ~ I will try to delete the duplicate comment~

jihoonson · 2018-10-19T06:02:21Z

I suspect a race problem of KafkaSupervisorTest. All tests in KafkaSupervisorTest basically executes the below code.

    supervisor.start();
    supervisor.runInternal();
    verifyAll();

However, supervisor.start() also internally calls supervisor.runInternal(). If this method is called twice or more, it can cause the unexpected method call errors reported above. This is not related to this PR anyway.

I left one question. Please take a look.

jihoonson

LGTM.

…afka causes the FullGC in Overlord (apache#6455) * remove consumer.listTopics() method * add consumerLock and exception handling for consumer.partitionFor() and remove some useless checks * add check in case consumer.partitionsFor() returns null * fix CI failure * fix failed UT * Revert "fix CI failure" This reverts commit f839d09. * revert unless commit and re-commit the useful part to fix failed UT

remove consumer.listTopics() method

b33a03f

elloooooo changed the title ~~Remove consumer.listTopics() method in case when too many topic in kafka causes the FullGC in Overlord~~ Remove consumer.listTopics() method in case when too many topics in kafka causes the FullGC in Overlord Oct 12, 2018

clintropolis requested changes Oct 12, 2018

View reviewed changes

add consumerLock and exception handling for consumer.partitionFor() a…

db750e0

…nd remove some useless checks

jihoonson reviewed Oct 15, 2018

View reviewed changes

add check in case consumer.partitionsFor() returns null

417834f

liuboyu added 4 commits October 17, 2018 13:37

fix CI failure

f839d09

fix failed UT

6196352

Revert "fix CI failure"

10a536c

This reverts commit f839d09.

revert unless commit and re-commit the useful part to fix failed UT

cd68430

jihoonson reviewed Oct 19, 2018

View reviewed changes

jihoonson approved these changes Oct 22, 2018

View reviewed changes

b-slim merged commit 1e82b62 into apache:master Oct 22, 2018

elloooooo deleted the feature-improve-get-topic branch October 26, 2018 02:02

jon-wei added this to the 0.14.0 milestone Feb 20, 2019

clintropolis mentioned this pull request May 6, 2019

Add errors and state to stream supervisor status API endpoint #7428

Merged

Conversation

elloooooo commented Oct 12, 2018

Uh oh!

clintropolis left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

b-slim commented Oct 13, 2018

Uh oh!

b-slim commented Oct 13, 2018

Uh oh!

elloooooo commented Oct 15, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

elloooooo Oct 15, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jihoonson commented Oct 15, 2018

Uh oh!

jihoonson commented Oct 15, 2018

Uh oh!

jihoonson commented Oct 15, 2018

Uh oh!

elloooooo commented Oct 18, 2018

Uh oh!

b-slim commented Oct 18, 2018

Uh oh!

gianm commented Oct 18, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jihoonson commented Oct 19, 2018

Uh oh!

jihoonson left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

elloooooo Oct 15, 2018 •

edited

Loading