{"action":"created","issue":{"url":"https://api.github.com/repos/apache/kafka/issues/5221","repository_url":"https://api.github.com/repos/apache/kafka","labels_url":"https://api.github.com/repos/apache/kafka/issues/5221/labels{/name}","comments_url":"https://api.github.com/repos/apache/kafka/issues/5221/comments","events_url":"https://api.github.com/repos/apache/kafka/issues/5221/events","html_url":"https://github.com/apache/kafka/pull/5221","id":332224716,"node_id":"MDExOlB1bGxSZXF1ZXN0MTk0NzYwMjE2","number":5221,"title":"KAFKA-7019 - make reading metadata lock-free by maintaining an atomically-updated read snapshot","user":{"login":"radai-rosenblatt","id":1042632,"node_id":"MDQ6VXNlcjEwNDI2MzI=","avatar_url":"https://avatars2.githubusercontent.com/u/1042632?v=4","gravatar_id":"","url":"https://api.github.com/users/radai-rosenblatt","html_url":"https://github.com/radai-rosenblatt","followers_url":"https://api.github.com/users/radai-rosenblatt/followers","following_url":"https://api.github.com/users/radai-rosenblatt/following{/other_user}","gists_url":"https://api.github.com/users/radai-rosenblatt/gists{/gist_id}","starred_url":"https://api.github.com/users/radai-rosenblatt/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/radai-rosenblatt/subscriptions","organizations_url":"https://api.github.com/users/radai-rosenblatt/orgs","repos_url":"https://api.github.com/users/radai-rosenblatt/repos","events_url":"https://api.github.com/users/radai-rosenblatt/events{/privacy}","received_events_url":"https://api.github.com/users/radai-rosenblatt/received_events","type":"User","site_admin":false},"labels":[{"id":838330122,"node_id":"MDU6TGFiZWw4MzgzMzAxMjI=","url":"https://api.github.com/repos/apache/kafka/labels/core","name":"core","color":"ed5a86","default":false},{"id":970614414,"node_id":"MDU6TGFiZWw5NzA2MTQ0MTQ=","url":"https://api.github.com/repos/apache/kafka/labels/performance","name":"performance","color":"c2e0c6","default":false}],"state":"open","locked":false,"assignee":{"login":"lindong28","id":5873403,"node_id":"MDQ6VXNlcjU4NzM0MDM=","avatar_url":"https://avatars0.githubusercontent.com/u/5873403?v=4","gravatar_id":"","url":"https://api.github.com/users/lindong28","html_url":"https://github.com/lindong28","followers_url":"https://api.github.com/users/lindong28/followers","following_url":"https://api.github.com/users/lindong28/following{/other_user}","gists_url":"https://api.github.com/users/lindong28/gists{/gist_id}","starred_url":"https://api.github.com/users/lindong28/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lindong28/subscriptions","organizations_url":"https://api.github.com/users/lindong28/orgs","repos_url":"https://api.github.com/users/lindong28/repos","events_url":"https://api.github.com/users/lindong28/events{/privacy}","received_events_url":"https://api.github.com/users/lindong28/received_events","type":"User","site_admin":false},"assignees":[{"login":"lindong28","id":5873403,"node_id":"MDQ6VXNlcjU4NzM0MDM=","avatar_url":"https://avatars0.githubusercontent.com/u/5873403?v=4","gravatar_id":"","url":"https://api.github.com/users/lindong28","html_url":"https://github.com/lindong28","followers_url":"https://api.github.com/users/lindong28/followers","following_url":"https://api.github.com/users/lindong28/following{/other_user}","gists_url":"https://api.github.com/users/lindong28/gists{/gist_id}","starred_url":"https://api.github.com/users/lindong28/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/lindong28/subscriptions","organizations_url":"https://api.github.com/users/lindong28/orgs","repos_url":"https://api.github.com/users/lindong28/repos","events_url":"https://api.github.com/users/lindong28/events{/privacy}","received_events_url":"https://api.github.com/users/lindong28/received_events","type":"User","site_admin":false}],"milestone":null,"comments":15,"created_at":"2018-06-14T02:01:01Z","updated_at":"2018-08-06T17:13:39Z","closed_at":null,"author_association":"CONTRIBUTOR","pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/5221","html_url":"https://github.com/apache/kafka/pull/5221","diff_url":"https://github.com/apache/kafka/pull/5221.diff","patch_url":"https://github.com/apache/kafka/pull/5221.patch"},"body":"we've seen cases where under a constant metadata update load (automated partition rebalance) request handling threads can block for a significant amount of time on the metadata lock.\r\n\r\nthe reason is that for large enough clusters (~200,000 topic partitions) a read operation can actually take a long while to compose a response. under a constant stream of reads + writes we see situations where a reader is currently in, a writer is pending (blocked) and then a big pile-up of more readers that are blocked behind the pending writer.\r\n\r\nthis patch makes the read path lock-free. the metadata is now stored in a logically-immutable snapshot. all read operations grab a snapshot and serve data out of it. write paths create an entirely new snapshot and atomically assign it. writers are still under a lock, for mutual exclusion.\r\n\r\nhere's the benchmark code i used to measure the effects of this patch:\r\n```java\r\npublic class MetadataCacheBenchmark {\r\n    private volatile boolean running = true;\r\n\r\n    int numBrokers = 150;\r\n    int numTopics = 3500;\r\n    int maxPartitionsPerTopic = 100;\r\n    int replicationFactor = 2;\r\n    int numUpdaters = 1;\r\n    double updateRateLimit = 10.0; //qps\r\n    int numReaders = 5;\r\n    boolean partialUpdate = true;\r\n\r\n    private final ListenerName listener = new ListenerName(\"listener\");\r\n\r\n    private final AtomicLong updateCounter = new AtomicLong();\r\n    private final AtomicLong readCounter = new AtomicLong();\r\n\r\n    @Test\r\n    public void benchmarkAllTheThings() throws Exception {\r\n        //long seed = System.currentTimeMillis();\r\n        long seed = 666;\r\n        System.err.println(\"seed is \" + seed);\r\n        Random r = new Random(seed);\r\n\r\n        MetadataCache cache = new MetadataCache(666);\r\n        UpdateMetadataRequest fullRequest = buildRequest(r, -1);\r\n        UpdateMetadataRequest partialRequest = buildRequest(r, 1);\r\n\r\n        cache.updateCache(0, fullRequest); //initial data (useful in case there are no writers)\r\n\r\n        Set<String> allTopics = new HashSet<>();\r\n        for (int i = 0; i < numTopics; i++) {\r\n            allTopics.add(\"topic-\" + i);\r\n        }\r\n        scala.collection.mutable.Set<String> topicsScalaSet = JavaConverters.asScalaSetConverter(allTopics).asScala();\r\n\r\n        Thread.UncaughtExceptionHandler exceptionHandler = new Thread.UncaughtExceptionHandler() {\r\n            @Override\r\n            public void uncaughtException(Thread t, Throwable e) {\r\n                running = false;\r\n                System.err.println(\"thread \" + t + \" died\");\r\n                e.printStackTrace(System.err);\r\n                System.exit(1);\r\n            }\r\n        };\r\n        List<Thread> threads = new ArrayList<>();\r\n\r\n        for (int i = 0; i < numUpdaters; i++) {\r\n            UpdateMetadataRequest req = partialUpdate ? partialRequest : fullRequest;\r\n\r\n            Runnable updaterRunnable;\r\n            if (updateRateLimit > 0) {\r\n                updaterRunnable = new RateLimitedUpdateRunnable(updateRateLimit, cache, req);\r\n            } else {\r\n                updaterRunnable = new UpdateRunnable(cache, req);\r\n            }\r\n            Thread updaterThread = new Thread(updaterRunnable, \"updater-\" + i);\r\n            updaterThread.setDaemon(true);\r\n            updaterThread.setUncaughtExceptionHandler(exceptionHandler);\r\n            threads.add(updaterThread);\r\n        }\r\n\r\n        for (int i = 0; i < numReaders; i++) {\r\n            ReadRunnable readRunnable = new ReadRunnable(cache, topicsScalaSet);\r\n            Thread readerThread = new Thread(readRunnable, \"reader-\" + i);\r\n            readerThread.setDaemon(true);\r\n            readerThread.setUncaughtExceptionHandler(exceptionHandler);\r\n            threads.add(readerThread);\r\n        }\r\n\r\n        for (Thread t : threads) {\r\n            t.start();\r\n        }\r\n\r\n        long prevTime = System.currentTimeMillis();\r\n        long prevUpdates = 0;\r\n        long prevReads = 0;\r\n\r\n        long now;\r\n        long updates;\r\n        long reads;\r\n\r\n        long timeDiff;\r\n        long updateDiff;\r\n        long readDiff;\r\n\r\n        double updateQps;\r\n        double readQps;\r\n\r\n        while (running) {\r\n            Thread.sleep(TimeUnit.SECONDS.toMillis(30));\r\n            now = System.currentTimeMillis();\r\n            updates = updateCounter.longValue();\r\n            reads = readCounter.longValue();\r\n\r\n            timeDiff = now - prevTime;\r\n            updateDiff = updates - prevUpdates;\r\n            readDiff = reads - prevReads;\r\n\r\n            updateQps = ((double) updateDiff * 1000) / timeDiff;\r\n            readQps = ((double) readDiff * 1000) / timeDiff;\r\n\r\n            prevTime = now;\r\n            prevUpdates = updates;\r\n            prevReads = reads;\r\n\r\n            System.err.println(\"updates: \" + updateQps + \" / sec\");\r\n            System.err.println(\"reads: \" + readQps + \" / sec\");\r\n        }\r\n    }\r\n\r\n    private UpdateMetadataRequest buildRequest(Random random, int numTopicsOverride) {\r\n        int controllerEpoch = 0;\r\n        int totalPartitions = 0;\r\n        Set<UpdateMetadataRequest.Broker> liveBrokers = new HashSet<>();\r\n\r\n        for (int i = 0; i < numBrokers; i++) {\r\n            UpdateMetadataRequest.EndPoint endPoint =\r\n                new UpdateMetadataRequest.EndPoint(\"host-\" + i, 6666, SecurityProtocol.PLAINTEXT, listener);\r\n            UpdateMetadataRequest.Broker broker =\r\n                new UpdateMetadataRequest.Broker(i, Collections.singletonList(endPoint), \"rack-\" + i);\r\n            liveBrokers.add(broker);\r\n        }\r\n\r\n        Map<TopicPartition, UpdateMetadataRequest.PartitionState> partitions = new HashMap<>();\r\n\r\n        int topicCount = numTopicsOverride > 0 ? numTopicsOverride : numTopics;\r\n\r\n        for (int i = 0; i < topicCount; i++) {\r\n            String topicName = \"topic-\" + i;\r\n            int numPartitions = 1 + random.nextInt(maxPartitionsPerTopic);\r\n            for (int j = 0; j < numPartitions; j++) {\r\n                TopicPartition tp = new TopicPartition(topicName, j);\r\n                List<Integer> replicas = pick(replicationFactor, numBrokers, random);\r\n                UpdateMetadataRequest.PartitionState state =\r\n                    new UpdateMetadataRequest.PartitionState(controllerEpoch, replicas.get(0), 0, replicas, 0, replicas,\r\n                        Collections.<Integer>emptyList());\r\n                partitions.put(tp, state);\r\n            }\r\n            totalPartitions += numPartitions;\r\n        }\r\n\r\n        UpdateMetadataRequest.Builder builder =\r\n            new UpdateMetadataRequest.Builder((short) 4, 0, controllerEpoch, partitions, liveBrokers);\r\n\r\n        UpdateMetadataRequest request = builder.build((short) 4);\r\n        System.err.println(\"request has \" + totalPartitions + \" TPs total\");\r\n        return request;\r\n    }\r\n\r\n    private List<Integer> pick(int howMany, int from, Random random) {\r\n        List<Integer> result = new ArrayList<>(howMany);\r\n        while (result.size() < howMany) {\r\n            int chosen = random.nextInt(from); //exclusive\r\n            if (!result.contains(chosen)) {\r\n                result.add(chosen);\r\n            }\r\n        }\r\n        return result;\r\n    }\r\n\r\n    private class UpdateRunnable implements Runnable {\r\n        private final MetadataCache cache;\r\n        private final UpdateMetadataRequest request;\r\n        private int counter = 0;\r\n\r\n        public UpdateRunnable(MetadataCache cache, UpdateMetadataRequest request) {\r\n            this.cache = cache;\r\n            this.request = request;\r\n        }\r\n\r\n        @Override\r\n        public void run() {\r\n            while (running) {\r\n                cache.updateCache(counter++, request);\r\n                updateCounter.incrementAndGet();\r\n            }\r\n        }\r\n    }\r\n\r\n    private class RateLimitedUpdateRunnable implements Runnable {\r\n        private final MetadataCache cache;\r\n        private final UpdateMetadataRequest request;\r\n        private int counter = 0;\r\n\r\n        private final double targetQps;\r\n        private final int intervalMillis;\r\n\r\n        public RateLimitedUpdateRunnable(double targetQps, MetadataCache cache, UpdateMetadataRequest request) {\r\n            this.cache = cache;\r\n            this.request = request;\r\n            this.targetQps = targetQps;\r\n            this.intervalMillis = (int) (1000.0 / targetQps);\r\n        }\r\n\r\n        @Override\r\n        public void run() {\r\n            while (running) {\r\n                long start = System.currentTimeMillis();\r\n                cache.updateCache(counter++, request);\r\n                long end = System.currentTimeMillis();\r\n                updateCounter.incrementAndGet();\r\n                long took = end - start;\r\n                long remaining = intervalMillis - took;\r\n                if (remaining > 0) {\r\n                    try {\r\n                        Thread.sleep(remaining);\r\n                    } catch (Exception e) {\r\n                        e.printStackTrace(System.err);\r\n                    }\r\n                }\r\n            }\r\n        }\r\n    }\r\n\r\n    private class ReadRunnable implements Runnable {\r\n        private final MetadataCache cache;\r\n        private final scala.collection.mutable.Set<String> topics;\r\n\r\n        public ReadRunnable(MetadataCache cache, scala.collection.mutable.Set<String> topics) {\r\n            this.cache = cache;\r\n            this.topics = topics;\r\n        }\r\n\r\n        @Override\r\n        public void run() {\r\n            while (running) {\r\n                cache.getTopicMetadata(topics, listener, false);\r\n                readCounter.incrementAndGet();\r\n            }\r\n        }\r\n    }\r\n}\r\n```\r\n\r\nthe interesting/problematic scenario is a combination of writers and readers,"},"comment":{"url":"https://api.github.com/repos/apache/kafka/issues/comments/410781612","html_url":"https://github.com/apache/kafka/pull/5221#issuecomment-410781612","issue_url":"https://api.github.com/repos/apache/kafka/issues/5221","id":410781612,"node_id":"MDEyOklzc3VlQ29tbWVudDQxMDc4MTYxMg==","user":{"login":"ijuma","id":24747,"node_id":"MDQ6VXNlcjI0NzQ3","avatar_url":"https://avatars2.githubusercontent.com/u/24747?v=4","gravatar_id":"","url":"https://api.github.com/users/ijuma","html_url":"https://github.com/ijuma","followers_url":"https://api.github.com/users/ijuma/followers","following_url":"https://api.github.com/users/ijuma/following{/other_user}","gists_url":"https://api.github.com/users/ijuma/gists{/gist_id}","starred_url":"https://api.github.com/users/ijuma/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ijuma/subscriptions","organizations_url":"https://api.github.com/users/ijuma/orgs","repos_url":"https://api.github.com/users/ijuma/repos","events_url":"https://api.github.com/users/ijuma/events{/privacy}","received_events_url":"https://api.github.com/users/ijuma/received_events","type":"User","site_admin":false},"created_at":"2018-08-06T17:13:39Z","updated_at":"2018-08-06T17:13:39Z","author_association":"CONTRIBUTOR","body":"@radai-rosenblatt Good point. `AnyRefMap` where the key is a reference and `LongMap` for int/long keys take a `initialBufferSize`. They seem like good candidates for this class. What do you think?"}}
{"actor":{"display_login":"ijuma"}}