GitHub PullRequestReviewCommentEvent at 2014-05-05 20:13:29
Repository: voltdb
Actor: akhanzode
Payload: {"comment":{"url":"https://api.github.com/repos/VoltDB/voltdb/pulls/comments/12297152","id":12297152,"diff_hunk":"@@ -120,123 +152,58 @@ public void validate() {\n          */\n         @Override\n         public void printUsage() {\n-            System.out\n-                    .println(\"Usage: kafkaloader [args] -t topic tablename\");\n+            System.out.println(\"Usage: kafkaloader [args] -z kafka-zookeeper -t topic tablename\");\n             super.printUsage();\n         }\n     }\n \n-    private final static AtomicLong failedCount = new AtomicLong(0);\n-\n-    static class KafkaBulkLoaderCallback implements BulkLoaderFailureCallBack {\n+    class KafkaBulkLoaderCallback implements BulkLoaderFailureCallBack {\n \n         @Override\n         public void failureCallback(Object rowHandle, Object[] fieldList, ClientResponse response) {\n             if (response.getStatus() != ClientResponse.SUCCESS) {\n-                long fc = failedCount.incrementAndGet();\n-                if (config.maxerrors > 0 && fc > config.maxerrors) {\n+                m_log.error(\"Failed to Insert Row: \" + rowHandle);\n+                long fc = m_failedCount.incrementAndGet();\n+                if (m_config.maxerrors > 0 && fc > m_config.maxerrors) {\n                     System.exit(1);\n                 }\n             }\n         }\n-\n     }\n \n     private static class KafkaConsumerConnector {\n \n-        final String m_host;\n-        final String m_port;\n-        ConsumerConfig consumerConfig;\n-        ConsumerConnector consumer;\n+        final ConsumerConfig m_consumerConfig;\n+        final ConsumerConnector m_consumer;\n \n-        public KafkaConsumerConnector(String host, String port) {\n-            m_host = host;\n-            m_port = port;\n-        }\n-\n-        public void buildConfig(String a_zookeeper) {\n-            //Should get this from properties file or something.\n+        public KafkaConsumerConnector(String zk, String tableName) {\n+            //Get group id which should be unique for table so as to keep offsets clean for multiple runs.\n+            String groupId = \"voltdb-\" + tableName;\n+            //TODO: Should get this from properties file or something as override?\n             Properties props = new Properties();\n-            props.put(\"zookeeper.connect\", a_zookeeper);\n-            props.put(\"group.id\", \"exportverifier\");\n+            props.put(\"zookeeper.connect\", zk);\n+            props.put(\"group.id\", groupId);\n             props.put(\"zookeeper.session.timeout.ms\", \"400\");\n             props.put(\"zookeeper.sync.time.ms\", \"200\");\n             props.put(\"auto.commit.interval.ms\", \"1000\");\n             props.put(\"auto.commit.enable\", \"true\");\n             props.put(\"auto.offset.reset\", \"smallest\");\n \n-            consumerConfig = new ConsumerConfig(props);\n+            m_consumerConfig = new ConsumerConfig(props);\n \n-            consumer = kafka.consumer.Consumer.createJavaConsumerConnector(consumerConfig);\n+            m_consumer = kafka.consumer.Consumer.createJavaConsumerConnector(m_consumerConfig);\n         }\n \n         public void stop() {\n-            consumer.commitOffsets();\n-            consumer.shutdown();\n-        }\n-    }\n-\n-\n-    /**\n-     * kafkaloader main\n-     *\n-     * @param args\n-     * @throws IOException\n-     * @throws InterruptedException\n-     *\n-     */\n-    public static void main(String[] args) {\n-\n-        final KafkaConfig cfg = new KafkaConfig();\n-        cfg.parse(KafkaLoader.class.getName(), args);\n-\n-        config = cfg;\n-\n-        // Split server list\n-        final String[] serverlist = config.servers.split(\",\");\n-\n-        // Create connection\n-        final ClientConfig c_config = new ClientConfig(config.user, config.password);\n-        c_config.setProcedureCallTimeout(0); // Set procedure all to infinite\n-        Client csvClient = null;\n-        VoltBulkLoader loader = null;\n-        try {\n-            csvClient = KafkaLoader.getClient(c_config, serverlist);\n-            loader = csvClient.getNewBulkLoader(config.table, config.batch, new KafkaBulkLoaderCallback());\n-        } catch (Exception e) {\n-            m_log.error(\"Error connecting to the servers: \"\n-                    + config.servers);\n-            System.exit(-1);\n-        }\n-\n-\n-        ExecutorService es = null;\n-        assert (csvClient != null);\n-        assert (loader != null);\n-        try {\n-            es = getConsumerExecutor(loader);\n-            es.awaitTermination(365, TimeUnit.DAYS);\n-        } catch (Exception ex) {\n-            m_log.error(\"Error in getting Kafka Consumer\", ex);\n-        }\n-\n-        Runtime.getRuntime().addShutdownHook(\n-                new Thread() {\n-                    @Override\n-                    public void run() {\n-                        for (KafkaConsumerConnector consumerConnector : m_consumers) {\n-                            consumerConnector.stop();\n-                        }\n-                    }\n-                });\n-\n-        try {\n-            loader.drain();\n-            loader.close();\n-        } catch (InterruptedException ex) {\n+            try {\n+                //Let offset get pushed to zk....so sleep for auto.commit.interval.ms\n+                Thread.sleep(1100);","path":"src/frontend/org/voltdb/utils/KafkaLoader.java","position":249,"original_position":249,"commit_id":"f7af8bddb875ae5d706c753eafcace7119fc6fe2","original_commit_id":"f7af8bddb875ae5d706c753eafcace7119fc6fe2","user":{"login":"akhanzode","id":1675698,"avatar_url":"https://avatars.githubusercontent.com/u/1675698?","gravatar_id":"f344d8d642daa16f661eff68e5e5b819","url":"https://api.github.com/users/akhanzode","html_url":"https://github.com/akhanzode","followers_url":"https://api.github.com/users/akhanzode/followers","following_url":"https://api.github.com/users/akhanzode/following{/other_user}","gists_url":"https://api.github.com/users/akhanzode/gists{/gist_id}","starred_url":"https://api.github.com/users/akhanzode/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/akhanzode/subscriptions","organizations_url":"https://api.github.com/users/akhanzode/orgs","repos_url":"https://api.github.com/users/akhanzode/repos","events_url":"https://api.github.com/users/akhanzode/events{/privacy}","received_events_url":"https://api.github.com/users/akhanzode/received_events","type":"User","site_admin":false},"body":"The offsets for consumers are saved in zk and interval is 1 sec so at exit to save offsets we sleep > 1s otherwise end up getting messages from previous saved offsets.","created_at":"2014-05-05T20:13:29Z","updated_at":"2014-05-05T20:13:29Z","html_url":"https://github.com/VoltDB/voltdb/pull/1345#discussion_r12297152","pull_request_url":"https://api.github.com/repos/VoltDB/voltdb/pulls/1345","_links":{"self":{"href":"https://api.github.com/repos/VoltDB/voltdb/pulls/comments/12297152"},"html":{"href":"https://github.com/VoltDB/voltdb/pull/1345#discussion_r12297152"},"pull_request":{"href":"https://api.github.com/repos/VoltDB/voltdb/pulls/1345"}}}}