{"action":"created","comment":{"url":"https://api.github.com/repos/apache/kafka/pulls/comments/320945405","pull_request_review_id":283857119,"id":320945405,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDMyMDk0NTQwNQ==","diff_hunk":"@@ -0,0 +1,610 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more\n+ * contributor license agreements. See the NOTICE file distributed with\n+ * this work for additional information regarding copyright ownership.\n+ * The ASF licenses this file to You under the Apache License, Version 2.0\n+ * (the \"License\"); you may not use this file except in compliance with\n+ * the License. You may obtain a copy of the License at\n+ *\n+ *    http://www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software\n+ * distributed under the License is distributed on an \"AS IS\" BASIS,\n+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n+ * See the License for the specific language governing permissions and\n+ * limitations under the License.\n+ */\n+\n+package org.apache.kafka.trogdor.workload;\n+\n+import com.fasterxml.jackson.annotation.JsonCreator;\n+import com.fasterxml.jackson.annotation.JsonProperty;\n+import com.fasterxml.jackson.databind.JsonNode;\n+import org.apache.kafka.clients.admin.AdminClient;\n+import org.apache.kafka.clients.consumer.ConsumerConfig;\n+import org.apache.kafka.clients.consumer.KafkaConsumer;\n+import org.apache.kafka.clients.producer.KafkaProducer;\n+import org.apache.kafka.clients.producer.ProducerConfig;\n+import org.apache.kafka.clients.producer.ProducerRecord;\n+import org.apache.kafka.common.TopicPartition;\n+import org.apache.kafka.common.internals.KafkaFutureImpl;\n+import org.apache.kafka.common.serialization.ByteArrayDeserializer;\n+import org.apache.kafka.common.serialization.ByteArraySerializer;\n+import org.apache.kafka.common.utils.Utils;\n+import org.apache.kafka.trogdor.common.JsonUtil;\n+import org.apache.kafka.trogdor.common.Platform;\n+import org.apache.kafka.trogdor.common.ThreadUtils;\n+import org.apache.kafka.trogdor.common.WorkerUtils;\n+import org.apache.kafka.trogdor.task.TaskWorker;\n+import org.apache.kafka.trogdor.task.WorkerStatusTracker;\n+import org.slf4j.Logger;\n+import org.slf4j.LoggerFactory;\n+\n+import java.time.Duration;\n+import java.util.ArrayList;\n+import java.util.Collections;\n+import java.util.Iterator;\n+import java.util.List;\n+import java.util.Properties;\n+import java.util.Random;\n+import java.util.UUID;\n+import java.util.concurrent.ExecutorService;\n+import java.util.concurrent.Executors;\n+import java.util.concurrent.Future;\n+import java.util.concurrent.ScheduledExecutorService;\n+import java.util.concurrent.TimeUnit;\n+import java.util.concurrent.atomic.AtomicBoolean;\n+import java.util.stream.Collectors;\n+\n+public class SustainedConnectionWorker implements TaskWorker {\n+    private static final Logger log = LoggerFactory.getLogger(SustainedConnectionWorker.class);\n+\n+    // This is the metadata for the test itself.\n+    private final String id;\n+    private final SustainedConnectionSpec spec;\n+\n+    // These variables are used to maintain the connections.\n+    private static final int THROTTLE_PERIOD_MS = 10;\n+    private ExecutorService workerExecutor;\n+    private final AtomicBoolean running = new AtomicBoolean(false);\n+    private KafkaFutureImpl<String> doneFuture;\n+    private ArrayList<SustainedConnection> connections;\n+\n+    // These variables are used when tracking the reported status of the worker.\n+    private static final int REPORT_INTERVAL_MS = 5000;\n+    private WorkerStatusTracker status;\n+    private long totalProducerConnections;\n+    private long totalProducerFailedConnections;\n+    private long totalConsumerConnections;\n+    private long totalConsumerFailedConnections;\n+    private long totalMetadataConnections;\n+    private long totalMetadataFailedConnections;\n+    private long totalAborts;\n+    private Future<?> statusUpdaterFuture;\n+    private ScheduledExecutorService statusUpdaterExecutor;\n+\n+    public SustainedConnectionWorker(String id, SustainedConnectionSpec spec) {\n+        this.id = id;\n+        this.spec = spec;\n+    }\n+\n+    @Override\n+    public void start(Platform platform, WorkerStatusTracker status,\n+                      KafkaFutureImpl<String> doneFuture) throws Exception {\n+        if (!running.compareAndSet(false, true)) {\n+            throw new IllegalStateException(\"SustainedConnectionWorker is already running.\");\n+        }\n+        log.info(\"{}: Activating SustainedConnectionWorker with {}\", id, spec);\n+        this.doneFuture = doneFuture;\n+        this.status = status;\n+        this.connections = new ArrayList<SustainedConnection>();\n+        synchronized (SustainedConnectionWorker.this) {\n+\n+            // Initialize all status reporting metrics to 0.\n+            this.totalProducerConnections = 0;\n+            this.totalProducerFailedConnections = 0;\n+            this.totalConsumerConnections = 0;\n+            this.totalConsumerFailedConnections = 0;\n+            this.totalMetadataConnections = 0;\n+            this.totalMetadataFailedConnections = 0;\n+            this.totalAborts = 0;\n+\n+            // Create the worker classes and add them to the list of items to act on.\n+            for (int i = 0; i < spec.producerConnectionCount(); i++) {\n+                this.connections.add(new ProducerSustainedConnection(this.spec));\n+            }\n+            for (int i = 0; i < spec.consumerConnectionCount(); i++) {\n+                this.connections.add(new ConsumerSustainedConnection(this.spec));\n+            }\n+            for (int i = 0; i < spec.metadataConnectionCount(); i++) {\n+                this.connections.add(new MetadataSustainedConnection(this.spec));\n+            }\n+        }\n+\n+        // Create the status reporter thread and schedule it.\n+        this.statusUpdaterExecutor = Executors.newScheduledThreadPool(1,\n+            ThreadUtils.createThreadFactory(\"StatusUpdaterWorkerThread%d\", false));\n+        this.statusUpdaterFuture = this.statusUpdaterExecutor.scheduleAtFixedRate(\n+            new StatusUpdater(), 0, REPORT_INTERVAL_MS, TimeUnit.MILLISECONDS);\n+\n+        // Create the maintainer pool, add all the maintainer threads, then start it.\n+        this.workerExecutor = Executors.newFixedThreadPool(spec.numThreads(),\n+            ThreadUtils.createThreadFactory(\"SustainedConnectionWorkerThread%d\", false));\n+        for (int i = 0; i < spec.numThreads(); i++) {\n+            this.workerExecutor.submit(new MaintainLoop());\n+        }\n+    }\n+\n+    private interface SustainedConnection extends AutoCloseable {\n+        boolean needsRefresh();\n+        void refresh();\n+        void claim();\n+    }\n+\n+    private class MetadataSustainedConnection implements SustainedConnection {\n+\n+        // This variable is used to maintain the connection itself.\n+        private AdminClient client;\n+\n+        // These variables are used to maintain the scheduling of the connection keep-alive.\n+        private long nextUpdate;\n+        private boolean inUse;\n+        private long refreshRate;\n+\n+        // This variable is used to maintain the connection properties.\n+        private final Properties props;\n+\n+        MetadataSustainedConnection(SustainedConnectionSpec spec) {\n+            // This variable is used to maintain the connection itself.\n+            this.client = null;\n+\n+            // These variables are used to maintain the scheduling of the connection keep-alive.\n+            this.nextUpdate = 0;\n+            this.inUse = false;\n+            this.refreshRate = spec.refreshRateMs();\n+\n+            // This variable is used to maintain the connection properties.\n+            this.props = new Properties();\n+            this.props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers());\n+            WorkerUtils.addConfigsToProperties(this.props, spec.commonClientConf(), spec.commonClientConf());\n+        }\n+\n+        @Override\n+        public void refresh() {\n+            try {\n+                if (this.client == null) {\n+\n+                    // Housekeeping to track the number of opened connections.\n+                    synchronized (SustainedConnectionWorker.this) {\n+                        SustainedConnectionWorker.this.totalMetadataConnections += 1;\n+                    }\n+\n+                    // Create the admin client connection.\n+                    this.client = AdminClient.create(this.props);\n+                }\n+\n+                // Fetch some metadata to keep the connection alive.\n+                this.client.describeCluster().nodes().get();\n+\n+            } catch (Throwable e) {\n+\n+                // Housekeeping to track the number of opened connections and failed connection attempts.\n+                synchronized (SustainedConnectionWorker.this) {\n+                    SustainedConnectionWorker.this.totalMetadataConnections -= 1;\n+                    SustainedConnectionWorker.this.totalMetadataFailedConnections += 1;\n+                }\n+\n+                SustainedConnectionWorker.log.error(e.getMessage());\n+\n+                // Set the admin client to be recreated on the next cycle.\n+                this.closeQuietly();\n+            }\n+\n+            // Schedule this again and set to not in use.\n+            this.nextUpdate = System.currentTimeMillis() + this.refreshRate;\n+            this.inUse = false;\n+        }\n+\n+        @Override\n+        public boolean needsRefresh() {\n+            return !this.inUse && (System.currentTimeMillis() > this.nextUpdate);\n+        }\n+\n+        @Override\n+        public void claim() {\n+            this.inUse = true;\n+        }\n+\n+        @Override\n+        public void close() throws Exception {\n+            closeQuietly();\n+        }\n+\n+        private void closeQuietly() {\n+            Utils.closeQuietly(this.client, \"AdminClient\");\n+            this.client = null;\n+        }\n+    }\n+\n+    private class ProducerSustainedConnection implements SustainedConnection {\n+\n+        // These variables are used to maintain the connection itself.\n+        private KafkaProducer<byte[], byte[]> producer;\n+        private List<TopicPartition> partitions;\n+        private String topicName;\n+        private Iterator<TopicPartition> partitionsIterator;\n+        private PayloadIterator keys;\n+        private PayloadIterator values;\n+\n+        // These variables are used to maintain the scheduling of the connection keep-alive.\n+        private long nextUpdate;\n+        private boolean inUse;\n+        private long refreshRate;\n+\n+        // This variable is used to maintain the connection properties.\n+        private final Properties props;\n+\n+        ProducerSustainedConnection(SustainedConnectionSpec spec) {\n+\n+            // These variables are used to maintain the connection itself.\n+            this.producer = null;\n+            this.partitions = null;\n+            this.topicName = spec.topicName();\n+            this.partitionsIterator = null;\n+            this.keys = new PayloadIterator(spec.keyGenerator());\n+            this.values = new PayloadIterator(spec.valueGenerator());\n+\n+            // These variables are used to maintain the scheduling of the connection keep-alive.\n+            this.nextUpdate = 0;\n+            this.inUse = false;\n+            this.refreshRate = spec.refreshRateMs();\n+\n+            // This variable is used to maintain the connection properties.\n+            this.props = new Properties();\n+            this.props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers());\n+            this.props.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, 1);\n+            WorkerUtils.addConfigsToProperties(this.props, spec.commonClientConf(), spec.producerConf());\n+        }\n+\n+        @Override\n+        public void refresh() {\n+            try {\n+                if (this.producer == null) {\n+\n+                    // Housekeeping to track the number of opened connections.\n+                    synchronized (SustainedConnectionWorker.this) {\n+                        SustainedConnectionWorker.this.totalProducerConnections += 1;\n+                    }\n+\n+                    // Create the producer, fetch the specified topic's partitions and randomize them.\n+                    this.producer = new KafkaProducer<>(this.props, new ByteArraySerializer(), new ByteArraySerializer());\n+                    this.partitions = this.producer.partitionsFor(this.topicName).stream()\n+                             .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))\n+                             .collect(Collectors.toList());\n+                    Collections.shuffle(this.partitions);\n+                }\n+\n+                // Create a new iterator over the partitions if the current one doesn't exist or is exhausted.\n+                if (this.partitionsIterator == null || !this.partitionsIterator.hasNext()) {\n+                    this.partitionsIterator = this.partitions.iterator();\n+                }\n+\n+                // Produce a single record and send it synchronously.\n+                TopicPartition partition = this.partitionsIterator.next();\n+                ProducerRecord<byte[], byte[]> record = new ProducerRecord<>(\n+                        partition.topic(), partition.partition(), keys.next(), values.next());\n+                producer.send(record).get();\n+\n+            } catch (Throwable e) {\n+\n+                // Housekeeping to track the number of opened connections and failed connection attempts.\n+                synchronized (SustainedConnectionWorker.this) {\n+                    SustainedConnectionWorker.this.totalProducerConnections -= 1;\n+                    SustainedConnectionWorker.this.totalProducerFailedConnections += 1;\n+                }\n+\n+                SustainedConnectionWorker.log.error(e.getMessage());\n+\n+                // Set the producer to be recreated on the next cycle.\n+                this.closeQuietly();\n+            }\n+\n+            // Schedule this again and set to not in use.\n+            this.nextUpdate = System.currentTimeMillis() + this.refreshRate;\n+            this.inUse = false;\n+        }\n+\n+        @Override\n+        public boolean needsRefresh() {\n+            return !this.inUse && (System.currentTimeMillis() > this.nextUpdate);\n+        }\n+\n+        @Override\n+        public void claim() {\n+            this.inUse = true;\n+        }\n+\n+        @Override\n+        public void close() throws Exception {\n+            this.closeQuietly();\n+        }\n+\n+        private void closeQuietly() {\n+            Utils.closeQuietly(this.producer, \"KafkaProducer\");\n+            this.producer = null;\n+            this.partitions = null;\n+            this.partitionsIterator = null;\n+        }\n+    }\n+\n+    private class ConsumerSustainedConnection implements SustainedConnection {\n+\n+        // These variables are used to maintain the connection itself.\n+        private String topicName;\n+        private KafkaConsumer<byte[], byte[]> consumer;\n+        private String consumerGroup;\n+        private TopicPartition activePartition;\n+\n+        // These variables are used to maintain the scheduling of the connection keep-alive.\n+        private long nextUpdate;\n+        private boolean inUse;\n+        private long refreshRate;\n+\n+        // This variable is used to maintain the connection properties.\n+        private final Properties props;\n+\n+\n+        ConsumerSustainedConnection(SustainedConnectionSpec spec) {\n+            // These variables are used to maintain the connection itself.\n+            this.topicName = spec.topicName();\n+            this.consumer = null;\n+            this.consumerGroup = UUID.randomUUID().toString();\n+            this.activePartition = null;\n+\n+            // These variables are used to maintain the scheduling of the connection keep-alive.\n+            this.nextUpdate = 0;\n+            this.inUse = false;\n+            this.refreshRate = spec.refreshRateMs();\n+\n+            // This variable is used to maintain the connection properties.\n+            this.props = new Properties();\n+            this.props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, spec.bootstrapServers());\n+            this.props.put(ConsumerConfig.GROUP_ID_CONFIG, this.consumerGroup);\n+            this.props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, \"latest\");\n+            this.props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 100000);\n+            this.props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);\n+            this.props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 1024);\n+            WorkerUtils.addConfigsToProperties(this.props, spec.commonClientConf(), spec.consumerConf());\n+        }\n+\n+        @Override\n+        public void refresh() {\n+            try {\n+                if (this.consumer == null) {\n+\n+                    // Housekeeping to track the number of opened connections.\n+                    synchronized (SustainedConnectionWorker.this) {\n+                        SustainedConnectionWorker.this.totalConsumerConnections += 1;\n+                    }\n+\n+                    // Create the consumer and fetch the partitions for the specified topic.\n+                    this.consumer = new KafkaConsumer<>(this.props, new ByteArrayDeserializer(), new ByteArrayDeserializer());\n+                    List<TopicPartition> partitions = this.consumer.partitionsFor(this.topicName).stream()\n+                            .map(partitionInfo -> new TopicPartition(partitionInfo.topic(), partitionInfo.partition()))\n+                            .collect(Collectors.toList());\n+\n+                    // Select a random partition and assign it.\n+                    Random rand = new Random();\n+                    this.activePartition = partitions.get(rand.nextInt(partitions.size()));\n+                    this.consumer.assign(Collections.singletonList(this.activePartition));\n+                }\n+\n+                // The behavior when passing in an empty list is to seek to the end of all subscribed partitions.\n+                this.consumer.seekToEnd(Collections.emptyList());\n+\n+                // Poll to keep the connection alive, ignoring any records returned.\n+                this.consumer.poll(Duration.ofMillis(50));\n+\n+            } catch (Throwable e) {\n+\n+                // Housekeeping to track the number of opened connections and failed connection attempts.\n+                synchronized (SustainedConnectionWorker.this) {\n+                    SustainedConnectionWorker.this.totalConsumerConnections -= 1;\n+                    SustainedConnectionWorker.this.totalConsumerFailedConnections += 1;\n+                }\n+\n+                SustainedConnectionWorker.log.error(e.getMessage());\n+\n+                // Set the consumer to be recreated on the next cycle.\n+                this.closeQuietly();\n+            }\n+\n+            // Schedule this again and set to not in use.\n+            this.nextUpdate = System.currentTimeMillis() + this.refreshRate;\n+            this.inUse = false;\n+        }\n+\n+        @Override\n+        public boolean needsRefresh() {\n+            return !this.inUse && (System.currentTimeMillis() > this.nextUpdate);\n+        }\n+\n+        @Override\n+        public void claim() {\n+            this.inUse = true;\n+        }\n+\n+        @Override\n+        public void close() throws Exception {\n+            this.closeQuietly();\n+        }\n+\n+        private void closeQuietly() {\n+            Utils.closeQuietly(this.consumer, \"KafkaConsumer\");\n+            this.consumer = null;\n+            this.activePartition = null;\n+        }\n+    }\n+\n+    private static class SustainedConnectionThrottle extends Throttle {\n+        SustainedConnectionThrottle(int maxPerPeriod) {\n+            super(maxPerPeriod, THROTTLE_PERIOD_MS);\n+        }\n+    }\n+\n+    public class MaintainLoop implements Runnable {\n+        @Override\n+        public void run() {\n+            Throttle throttle = new SustainedConnectionThrottle(1);\n+            try {\n+                while (!doneFuture.isDone()) {\n+                    SustainedConnection currentConnection = SustainedConnectionWorker.this.findConnectionToMaintain();\n+                    if (currentConnection != null) {\n+                        currentConnection.refresh();\n+                    } else {\n+                        throttle.increment();\n+                    }\n+                }\n+            } catch (Exception e) {\n+                synchronized (SustainedConnectionWorker.this) {\n+                    SustainedConnectionWorker.this.totalAborts += 1;\n+                }\n+                WorkerUtils.abort(log, \"ConnectLoop\", e, doneFuture);\n+            }\n+        }\n+    }\n+\n+    private SustainedConnection findConnectionToMaintain() {\n+        synchronized (SustainedConnectionWorker.this) {\n+            for (SustainedConnection connection : this.connections) {\n+                if (connection.needsRefresh()) {\n+                    connection.claim();\n+                    return connection;\n+                }\n+            }\n+        }\n+        return null;\n+    }\n+\n+    private class StatusUpdater implements Runnable {\n+        @Override\n+        public void run() {\n+            try {\n+                JsonNode node = null;\n+                synchronized (SustainedConnectionWorker.this) {\n+                    node = JsonUtil.JSON_SERDE.valueToTree(\n+                        new StatusData(\n+                                SustainedConnectionWorker.this.totalProducerConnections,\n+                                SustainedConnectionWorker.this.totalProducerFailedConnections,\n+                                SustainedConnectionWorker.this.totalConsumerConnections,\n+                                SustainedConnectionWorker.this.totalConsumerFailedConnections,\n+                                SustainedConnectionWorker.this.totalMetadataConnections,\n+                                SustainedConnectionWorker.this.totalMetadataFailedConnections,\n+                                SustainedConnectionWorker.this.totalAborts,\n+                                System.currentTimeMillis()));\n+                }\n+                status.update(node);\n+            } catch (Exception e) {\n+                WorkerUtils.abort(log, \"StatusUpdater\", e, doneFuture);\n+            }\n+        }\n+    }\n+\n+    public static class StatusData {\n+        private final long totalProducerConnections;\n+        private final long totalProducerFailedConnections;\n+        private final long totalConsumerConnections;\n+        private final long totalConsumerFailedConnections;\n+        private final long totalMetadataConnections;\n+        private final long totalMetadataFailedConnections;\n+        private final long totalAborts;\n+        private final long updatedMs;\n+\n+        @JsonCreator\n+        StatusData(@JsonProperty(\"totalProducerConnections\") long totalProducerConnections,\n+                   @JsonProperty(\"totalProducerFailedConnections\") long totalProducerFailedConnections,\n+                   @JsonProperty(\"totalConsumerConnections\") long totalConsumerConnections,\n+                   @JsonProperty(\"totalConsumerFailedConnections\") long totalConsumerFailedConnections,\n+                   @JsonProperty(\"totalMetadataConnections\") long totalMetadataConnections,\n+                   @JsonProperty(\"totalMetadataFailedConnections\") long totalMetadataFailedConnections,\n+                   @JsonProperty(\"totalAborts\") long totalAborts,\n+                   @JsonProperty(\"updatedMs\") long updatedMs) {\n+            this.totalProducerConnections = totalProducerConnections;\n+            this.totalProducerFailedConnections = totalProducerFailedConnections;\n+            this.totalConsumerConnections = totalConsumerConnections;\n+            this.totalConsumerFailedConnections = totalConsumerFailedConnections;\n+            this.totalMetadataConnections = totalMetadataConnections;\n+            this.totalMetadataFailedConnections = totalMetadataFailedConnections;\n+            this.totalAborts = totalAborts;\n+            this.updatedMs = updatedMs;\n+        }\n+\n+        @JsonProperty\n+        public long totalProducerConnections() {\n+            return totalProducerConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalProducerFailedConnections() {\n+            return totalProducerFailedConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalConsumerConnections() {\n+            return totalConsumerConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalConsumerFailedConnections() {\n+            return totalConsumerFailedConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalMetadataConnections() {\n+            return totalMetadataConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalMetadataFailedConnections() {\n+            return totalMetadataFailedConnections;\n+        }\n+\n+        @JsonProperty\n+        public long totalAborts() {\n+            return totalAborts;\n+        }\n+\n+        @JsonProperty\n+        public long updatedMs() {\n+            return updatedMs;\n+        }\n+    }\n+\n+    @Override\n+    public void stop(Platform platform) throws Exception {\n+        if (!running.compareAndSet(true, false)) {\n+            throw new IllegalStateException(\"SustainedConnectionWorker is not running.\");\n+        }\n+        log.info(\"{}: Deactivating SustainedConnectionWorker.\", id);\n+\n+        // Shut down the periodic status updater and perform a final update on the\n+        // statistics.  We want to do this first, before deactivating any threads.\n+        // Otherwise, if some threads take a while to terminate, this could lead\n+        // to a misleading rate getting reported.\n+        this.statusUpdaterFuture.cancel(false);\n+        this.statusUpdaterExecutor.shutdown();\n+        this.statusUpdaterExecutor.awaitTermination(1, TimeUnit.HOURS);\n+        this.statusUpdaterExecutor = null;\n+        new StatusUpdater().run();","path":"tools/src/main/java/org/apache/kafka/trogdor/workload/SustainedConnectionWorker.java","position":598,"original_position":598,"commit_id":"5124f004fe794011a4f10c5fb253ceed349a9e76","original_commit_id":"5124f004fe794011a4f10c5fb253ceed349a9e76","user":{"login":"stanislavkozlovski","id":13639618,"node_id":"MDQ6VXNlcjEzNjM5NjE4","avatar_url":"https://avatars1.githubusercontent.com/u/13639618?v=4","gravatar_id":"","url":"https://api.github.com/users/stanislavkozlovski","html_url":"https://github.com/stanislavkozlovski","followers_url":"https://api.github.com/users/stanislavkozlovski/followers","following_url":"https://api.github.com/users/stanislavkozlovski/following{/other_user}","gists_url":"https://api.github.com/users/stanislavkozlovski/gists{/gist_id}","starred_url":"https://api.github.com/users/stanislavkozlovski/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/stanislavkozlovski/subscriptions","organizations_url":"https://api.github.com/users/stanislavkozlovski/orgs","repos_url":"https://api.github.com/users/stanislavkozlovski/repos","events_url":"https://api.github.com/users/stanislavkozlovski/events{/privacy}","received_events_url":"https://api.github.com/users/stanislavkozlovski/received_events","type":"User","site_admin":false},"body":"👍 ","created_at":"2019-09-04T19:54:33Z","updated_at":"2019-09-04T19:54:33Z","html_url":"https://github.com/apache/kafka/pull/7289#discussion_r320945405","pull_request_url":"https://api.github.com/repos/apache/kafka/pulls/7289","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments/320945405"},"html":{"href":"https://github.com/apache/kafka/pull/7289#discussion_r320945405"},"pull_request":{"href":"https://api.github.com/repos/apache/kafka/pulls/7289"}},"in_reply_to_id":320525394},"pull_request":{"url":"https://api.github.com/repos/apache/kafka/pulls/7289","id":313769137,"node_id":"MDExOlB1bGxSZXF1ZXN0MzEzNzY5MTM3","html_url":"https://github.com/apache/kafka/pull/7289","diff_url":"https://github.com/apache/kafka/pull/7289.diff","patch_url":"https://github.com/apache/kafka/pull/7289.patch","issue_url":"https://api.github.com/repos/apache/kafka/issues/7289","number":7289,"state":"open","locked":false,"title":"KAFKA-8853: Create sustained connections test for Trogdor","user":{"login":"scott-hendricks","id":49452084,"node_id":"MDQ6VXNlcjQ5NDUyMDg0","avatar_url":"https://avatars3.githubusercontent.com/u/49452084?v=4","gravatar_id":"","url":"https://api.github.com/users/scott-hendricks","html_url":"https://github.com/scott-hendricks","followers_url":"https://api.github.com/users/scott-hendricks/followers","following_url":"https://api.github.com/users/scott-hendricks/following{/other_user}","gists_url":"https://api.github.com/users/scott-hendricks/gists{/gist_id}","starred_url":"https://api.github.com/users/scott-hendricks/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/scott-hendricks/subscriptions","organizations_url":"https://api.github.com/users/scott-hendricks/orgs","repos_url":"https://api.github.com/users/scott-hendricks/repos","events_url":"https://api.github.com/users/scott-hendricks/events{/privacy}","received_events_url":"https://api.github.com/users/scott-hendricks/received_events","type":"User","site_admin":false},"body":"This creates a test that generates sustained connections against Kafka.  There\r\nare three different components we can stress with this, KafkaConsumer,\r\nKafkaProducer, and AdminClient.  This test tries use minimal bandwidth per\r\nconnection to reduce overhead impacts.\r\n\r\nThis test works by creating a threadpool that creates connections and then\r\nmaintains a central pool of connections at a specified keepalive rate.  The\r\nkeepalive action varies by which component is being stressed:\r\n\r\n  * KafkaProducer:  Sends one single produce record.  The configuration for\r\n    the produce request uses the same key/value generator as the ProduceBench\r\n    test.\r\n\r\n  * KafkaConsumer: Subscribes to a single partition, seeks to the end, and\r\n    then polls a minimal number of records.  Each consumer connection is its\r\n    own consumer group, and defaults to 1024 bytes as FETCH_MAX_BYTES to keep\r\n    traffic to a minimum.\r\n\r\n  * AdminClient: Makes an API call to get the nodes in the cluster.\r\n\r\nNOTE: This test is designed to be run alongside a ProduceBench test for a\r\nspecific topic, due to the way the Consumer test polls a single partition.\r\nThere may be no data returned by the consumer test if this is run on its own.\r\nThe connection should still be kept alive, but with no data returned.","created_at":"2019-09-03T21:13:20Z","updated_at":"2019-09-04T19:54:33Z","closed_at":null,"merged_at":null,"merge_commit_sha":"d7837a20735fa20d3f59b61fd4c35d4e27163ff5","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/kafka/pulls/7289/commits","review_comments_url":"https://api.github.com/repos/apache/kafka/pulls/7289/comments","review_comment_url":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/kafka/issues/7289/comments","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/5124f004fe794011a4f10c5fb253ceed349a9e76","head":{"label":"scott-hendricks:trunk","ref":"trunk","sha":"5124f004fe794011a4f10c5fb253ceed349a9e76","user":{"login":"scott-hendricks","id":49452084,"node_id":"MDQ6VXNlcjQ5NDUyMDg0","avatar_url":"https://avatars3.githubusercontent.com/u/49452084?v=4","gravatar_id":"","url":"https://api.github.com/users/scott-hendricks","html_url":"https://github.com/scott-hendricks","followers_url":"https://api.github.com/users/scott-hendricks/followers","following_url":"https://api.github.com/users/scott-hendricks/following{/other_user}","gists_url":"https://api.github.com/users/scott-hendricks/gists{/gist_id}","starred_url":"https://api.github.com/users/scott-hendricks/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/scott-hendricks/subscriptions","organizations_url":"https://api.github.com/users/scott-hendricks/orgs","repos_url":"https://api.github.com/users/scott-hendricks/repos","events_url":"https://api.github.com/users/scott-hendricks/events{/privacy}","received_events_url":"https://api.github.com/users/scott-hendricks/received_events","type":"User","site_admin":false},"repo":{"id":206123323,"node_id":"MDEwOlJlcG9zaXRvcnkyMDYxMjMzMjM=","name":"kafka","full_name":"scott-hendricks/kafka","private":false,"owner":{"login":"scott-hendricks","id":49452084,"node_id":"MDQ6VXNlcjQ5NDUyMDg0","avatar_url":"https://avatars3.githubusercontent.com/u/49452084?v=4","gravatar_id":"","url":"https://api.github.com/users/scott-hendricks","html_url":"https://github.com/scott-hendricks","followers_url":"https://api.github.com/users/scott-hendricks/followers","following_url":"https://api.github.com/users/scott-hendricks/following{/other_user}","gists_url":"https://api.github.com/users/scott-hendricks/gists{/gist_id}","starred_url":"https://api.github.com/users/scott-hendricks/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/scott-hendricks/subscriptions","organizations_url":"https://api.github.com/users/scott-hendricks/orgs","repos_url":"https://api.github.com/users/scott-hendricks/repos","events_url":"https://api.github.com/users/scott-hendricks/events{/privacy}","received_events_url":"https://api.github.com/users/scott-hendricks/received_events","type":"User","site_admin":false},"html_url":"https://github.com/scott-hendricks/kafka","description":"Mirror of Apache Kafka","fork":true,"url":"https://api.github.com/repos/scott-hendricks/kafka","forks_url":"https://api.github.com/repos/scott-hendricks/kafka/forks","keys_url":"https://api.github.com/repos/scott-hendricks/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/scott-hendricks/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/scott-hendricks/kafka/teams","hooks_url":"https://api.github.com/repos/scott-hendricks/kafka/hooks","issue_events_url":"https://api.github.com/repos/scott-hendricks/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/scott-hendricks/kafka/events","assignees_url":"https://api.github.com/repos/scott-hendricks/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/scott-hendricks/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/scott-hendricks/kafka/tags","blobs_url":"https://api.github.com/repos/scott-hendricks/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/scott-hendricks/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/scott-hendricks/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/scott-hendricks/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/scott-hendricks/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/scott-hendricks/kafka/languages","stargazers_url":"https://api.github.com/repos/scott-hendricks/kafka/stargazers","contributors_url":"https://api.github.com/repos/scott-hendricks/kafka/contributors","subscribers_url":"https://api.github.com/repos/scott-hendricks/kafka/subscribers","subscription_url":"https://api.github.com/repos/scott-hendricks/kafka/subscription","commits_url":"https://api.github.com/repos/scott-hendricks/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/scott-hendricks/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/scott-hendricks/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/scott-hendricks/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/scott-hendricks/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/scott-hendricks/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/scott-hendricks/kafka/merges","archive_url":"https://api.github.com/repos/scott-hendricks/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/scott-hendricks/kafka/downloads","issues_url":"https://api.github.com/repos/scott-hendricks/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/scott-hendricks/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/scott-hendricks/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/scott-hendricks/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/scott-hendricks/kafka/labels{/name}","releases_url":"https://api.github.com/repos/scott-hendricks/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/scott-hendricks/kafka/deployments","created_at":"2019-09-03T16:33:23Z","updated_at":"2019-09-03T20:59:38Z","pushed_at":"2019-09-03T20:59:35Z","git_url":"git://github.com/scott-hendricks/kafka.git","ssh_url":"git@github.com:scott-hendricks/kafka.git","clone_url":"https://github.com/scott-hendricks/kafka.git","svn_url":"https://github.com/scott-hendricks/kafka","homepage":null,"size":86946,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"disabled":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"trunk"}},"base":{"label":"apache:trunk","ref":"trunk","sha":"74fc3323a5815fad80318e96a88dc2164c700809","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":2211243,"node_id":"MDEwOlJlcG9zaXRvcnkyMjExMjQz","name":"kafka","full_name":"apache/kafka","private":false,"owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"html_url":"https://github.com/apache/kafka","description":"Mirror of Apache Kafka","fork":false,"url":"https://api.github.com/repos/apache/kafka","forks_url":"https://api.github.com/repos/apache/kafka/forks","keys_url":"https://api.github.com/repos/apache/kafka/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/kafka/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/kafka/teams","hooks_url":"https://api.github.com/repos/apache/kafka/hooks","issue_events_url":"https://api.github.com/repos/apache/kafka/issues/events{/number}","events_url":"https://api.github.com/repos/apache/kafka/events","assignees_url":"https://api.github.com/repos/apache/kafka/assignees{/user}","branches_url":"https://api.github.com/repos/apache/kafka/branches{/branch}","tags_url":"https://api.github.com/repos/apache/kafka/tags","blobs_url":"https://api.github.com/repos/apache/kafka/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/kafka/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/kafka/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/kafka/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/kafka/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/kafka/languages","stargazers_url":"https://api.github.com/repos/apache/kafka/stargazers","contributors_url":"https://api.github.com/repos/apache/kafka/contributors","subscribers_url":"https://api.github.com/repos/apache/kafka/subscribers","subscription_url":"https://api.github.com/repos/apache/kafka/subscription","commits_url":"https://api.github.com/repos/apache/kafka/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/kafka/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/kafka/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/kafka/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/kafka/contents/{+path}","compare_url":"https://api.github.com/repos/apache/kafka/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/kafka/merges","archive_url":"https://api.github.com/repos/apache/kafka/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/kafka/downloads","issues_url":"https://api.github.com/repos/apache/kafka/issues{/number}","pulls_url":"https://api.github.com/repos/apache/kafka/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/kafka/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/kafka/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/kafka/labels{/name}","releases_url":"https://api.github.com/repos/apache/kafka/releases{/id}","deployments_url":"https://api.github.com/repos/apache/kafka/deployments","created_at":"2011-08-15T18:06:16Z","updated_at":"2019-09-04T19:46:36Z","pushed_at":"2019-09-04T19:52:45Z","git_url":"git://github.com/apache/kafka.git","ssh_url":"git@github.com:apache/kafka.git","clone_url":"https://github.com/apache/kafka.git","svn_url":"https://github.com/apache/kafka","homepage":null,"size":86975,"stargazers_count":13444,"watchers_count":13444,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":7146,"mirror_url":null,"archived":false,"disabled":false,"open_issues_count":658,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":7146,"open_issues":658,"watchers":13444,"default_branch":"trunk"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/kafka/pulls/7289"},"html":{"href":"https://github.com/apache/kafka/pull/7289"},"issue":{"href":"https://api.github.com/repos/apache/kafka/issues/7289"},"comments":{"href":"https://api.github.com/repos/apache/kafka/issues/7289/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/kafka/pulls/7289/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/kafka/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/kafka/pulls/7289/commits"},"statuses":{"href":"https://api.github.com/repos/apache/kafka/statuses/5124f004fe794011a4f10c5fb253ceed349a9e76"}},"author_association":"NONE"}}
{"actor":{"display_login":"stanislavkozlovski"}}