Java pub-sub Implementation #135

rudy2steiner · 2019-01-27T16:02:40Z

What changes were proposed in this pull request?

provide a producer and consumer implement for confluo as a pub/sub system in java
also include a simple produce/consume benchmark test

How was this patch tested?

TestPubSub class provide unit tests

related issue #123 ,#125

…hmark merge

AmplabJenkins · 2019-01-27T16:02:41Z

Can one of the admins verify this patch?

anuragkh · 2019-01-27T20:41:32Z

Jenkins test this please.

AmplabJenkins · 2019-01-27T20:48:29Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/confluo-prb/360/
Test FAILed.

rudy2steiner · 2019-01-28T19:11:09Z

Jenkins test this please.

anuragkh · 2019-01-28T19:13:39Z

Jenkins test this please.

AmplabJenkins · 2019-01-28T19:19:20Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/confluo-prb/361/
Test PASSed.

anuragkh · 2019-01-28T19:15:53Z

javaclient/pom.xml

-          <descriptorRefs>
-            <descriptorRef>jar-with-dependencies</descriptorRef>
-          </descriptorRefs>
+          <!-- not append assembly id in release file name -->


Is there a reason why you switched from jar-with-dependencies to separate assembly?

i used to package all the dependency into lib,but it's ok to use jar-with-dependencies

anuragkh · 2019-01-28T19:16:28Z

javaclient/src/main/bin/start_consumer.sh

@@ -0,0 +1,42 @@
+#!/bin/sh


This file should be under javaclient/sbin instead.

i decided to remove the shell

anuragkh · 2019-01-28T19:16:38Z

javaclient/src/main/bin/start_producer.sh

@@ -0,0 +1,42 @@
+#!/bin/sh


Same as above.

anuragkh · 2019-01-28T19:20:38Z

javaclient/src/main/java/confluo/PropertiesParser.java

+     *  @return  filtered properties by domain
+     **/
+    public static Properties parse(String domain, String file) throws IOException{
+        URL url= PropertiesParser.class.getClassLoader().getResource(file);


The code formatting seems to be off (e.g., 2 spaces indentation, spaces before =, etc.). We use Google style for code formatting, please see here: https://github.com/google/google-java-format

have reformatted my code by google-java-format

anuragkh · 2019-01-28T19:23:48Z

javaclient/src/main/java/confluo/rpc/RpcClient.java

+   * @return  multilog id
+   * @throws TException Cannot get the atomic multilog
+   */
+  public long getAtomicMultilog(String name) throws TException {


Perhaps rename this to getAtomicMultiLogId()

anuragkh · 2019-01-28T19:24:21Z

javaclient/src/main/java/confluo/PropertiesParser.java

+public class PropertiesParser {
+
+    /**
+     *  parse unique domain properties


We use sentence case for documentation, i.e., capitalize first letter of each sentence. Please use this consistently across all documentation.

anuragkh · 2019-01-28T19:38:32Z

javaclient/src/main/java/confluo/rpc/MessageBatchBuilder.java

+
+/**
+ *
+ *  A builder for a batch of message,which different from


I still don't understand the need for a separate MessageBatchBuilder?

a test case(testBatchWriteReadInMemory) in TestRpcClient was added,which try to batch 3 records into MessageBatchBuilder,but can't read all the record from confluod, which may cause by MessageBatchBuilder.so i try to don't care the record timestamp,batch record seperately in MessageBatchBuilder
you may check this?

anuragkh · 2019-01-28T19:38:55Z

javaclient/src/main/java/confluo/rpc/RpcClient.java

@@ -237,14 +261,9 @@ public void removeTrigger(String triggerName) throws TException {
    client.removeTrigger(curMultilogId, triggerName);
  }

-  /**


Why did we remove this?

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java

anuragkh · 2019-01-28T19:41:07Z

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java

+    /**
+     *
+     **/
+    public void start() throws TException {


Shouldn't this be in the constructor?

javaclient/src/main/java/confluo/streaming/ConfluoProducer.java

anuragkh · 2019-01-28T19:45:19Z

javaclient/src/main/java/confluo/streaming/ConfluoProducer.java

+     * stop and close client
+     *
+     **/
+    public void stop() throws  TException{


A cleaner approach would be to extend Closeable, and implement this as close().

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java

javaclient/src/main/java/confluo/streaming/MessageListener.java

rudy2steiner · 2019-01-29T03:29:11Z

thanks for your code review, i'll try to optimize my implementation following your advices in next few days @anuragkh

anuragkh · 2019-01-29T03:52:04Z

Thanks for submitting the PR! Look forward to your changes.

rudy2steiner · 2019-02-13T16:22:08Z

@anuragkh you may review this PR again

anuragkh · 2019-02-19T09:50:01Z

javaclient/src/main/java/confluo/rpc/RecordBatchBuilder.java

@@ -27,6 +26,7 @@
  RecordBatchBuilder(Schema schema) {
    this.numRecords = 0;
    this.schema = schema;
+    this.batch=new TreeMap();


Formatting issue

anuragkh · 2019-02-19T09:50:28Z

javaclient/src/main/java/confluo/rpc/MessageBatchBuilder.java

+    public MessageBatchBuilder(Schema schema) {
+        this.numRecords = 0;
+        this.schema = schema;
+        this.batch= new rpc_record_batch();


Formatting issue

sorry for missing formate this class

javaclient/src/main/java/confluo/rpc/RpcClient.java

anuragkh · 2019-02-19T09:51:23Z

javaclient/src/main/java/confluo/rpc/RpcClient.java

+  public List<Record> readBatch(long offset,int batchSize) throws TException {
+    ByteBuffer batchBuffer=readBatchRaw(offset,batchSize);
+    List<Record> batchResult=new ArrayList<>(batchSize);
+    int remaining=batchBuffer.remaining();


Formatting issues in multiple locations in this file.

anuragkh · 2019-02-19T09:52:04Z

javaclient/src/main/java/confluo/rpc/Schema.java

+   * @param rec The record.
+   * @return The packed ByteBuffer.
+   */
+  public ByteBuffer pack(boolean direct,String... rec) {


Formatting issue

anuragkh · 2019-02-19T10:05:10Z

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java

+
+  public ConfluoConsumer(Properties properties) throws TException {
+
+    host = properties.getProperty(ConsumerConfig.BOOTSTRAP_ADDRESS, ConsumerConfig.BOOTSTRAP_ADDRESS_DEFAULT);


See comments on ProducerConfig

anuragkh · 2019-02-19T10:05:28Z

javaclient/src/main/java/confluo/rpc/RpcClient.java

+    return curSchema;
+  }
+
+


Unnecessary newlines

anuragkh · 2019-02-19T10:08:56Z

javaclient/src/main/java/confluo/readme.md

@@ -0,0 +1,10 @@
+


I think it would be better to move this readme to the confluo/streaming/ directory, since it does not apply to the entire confluo/ directory. Also, it would make the readme more readable with the following changes:
(1) There should be a heading; maybe Confluo Pub-sub?
(2) Use sentence case instead of using lower-case for the entire document.

anuragkh · 2019-02-19T10:09:28Z

javaclient/src/main/java/confluo/PropertiesParser.java

+   */
+  public static Properties parse(String domain, String file) throws IOException {
+    URL url = PropertiesParser.class.getClassLoader().getResource(file);
+    Properties pro = new Properties();


pro -> properties

anuragkh · 2019-02-19T10:09:58Z

javaclient/src/main/java/confluo/streaming/ConfluoProducer.java

+        batched++;
+        if (batched >= produceBatchSize) {
+          long offset = client.appendBatch(batchBuilder.getBatch());
+          // logger.info("write offset:"+offset);


Remove unused comment

anuragkh · 2019-02-19T10:12:08Z

Most of the changes look great! I left some minor comments on formatting, it will make the code much more readable if we can get those in.

haal · 2019-02-20T21:13:21Z

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java

+      batchSize = records.size();
+      messageListener.accept(records);
+    } else {
+      List<Record> messages=new ArrayList<>();


Can we init messages with client.read(offset) (like in prefetch case)?

sorry, i don't understand how to do

haal · 2019-02-20T21:18:38Z

javaclient/src/main/java/confluo/streaming/benchmark/ConsumerBenchmark.java

+      }
+      long actualDurationMs = System.currentTimeMillis() - start;
+      long qps = totalConsumedMsgNum * 1000 / actualDurationMs;
+      logger.error(


Is this info or error?

should be info ,thanks for you comment

haal · 2019-02-20T21:19:31Z

javaclient/src/main/java/confluo/streaming/benchmark/ConsumerBenchmark.java

+              "%d consumer finished,total msg:%d, elapsed:%d ms, qps:%d/s",
+              consumeNum, totalConsumedMsgNum, actualDurationMs, qps));
+    } catch (InterruptedException e) {
+      logger.info("interrupted", e);


Is this error logger.error ?

i mean to log the exception in INFO level

haal · 2019-02-20T21:20:38Z

javaclient/src/main/java/confluo/streaming/benchmark/ProducerBenchmark.java

+              "%d producer finished,total msg:%d, elapsed:%d ms, qps:%d/s",
+              producerNum, totalProducedMsgNum, actualDurationMs, qps));
+    } catch (InterruptedException e) {
+      logger.info("interrupted", e);


logger.error?

same as above

rudy2steiner · 2019-02-21T16:15:42Z

add ConfluoProducer/Cosumer pub-sub test,hope my changes make my PR more readable

anuragkh · 2019-03-07T04:02:09Z

Hi @rudy2steiner, sorry I was traveling for the past few days. I think this patch still needs better tests -- please see the relevant comments on the test.

rudy2steiner · 2019-03-08T09:50:03Z

removed non-unit test

anuragkh · 2019-03-12T16:35:01Z

Hi @rudy2steiner, thanks for that change. What I meant was we need to add tests that assert the correctness of the implementation. Currently the existing tests do not assert anything. We should update the tests to make sure we are checking for correctness using asserts.

rudy2steiner · 2019-03-14T02:27:02Z

Have you notice the major change in the class TestPubSub in the commit: 7f9e0cb;
And the newest TestPubSub.java contains the testConfluoPubSub function, in that i have 4 steps to assert the correctness of the implements(produce/consume) as following:

record the current offset,named mq_offset_mark, in target topic,eg mq_001
produce some of messages(pMeesages) to mq_001,
starting consume from the old offset mq_offset_mark
assert consume messages(cMessages) with pMessages in order

https://github.com/ucbrise/confluo/blob/7f9e0cbd41f625a9b37e6b0b7df035ccc74627f9/javaclient/src/test/java/TestPubSub.java

AmplabJenkins · 2021-03-16T03:50:54Z

Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/confluo-prb/387/

wangjin18 and others added 14 commits December 28, 2018 00:05

benchmark

34d1a34

batch produce and consume

2822908

reuse old topic

e5b50f6

confluo produce and consume duration

82ce485

confluo produce benchmark

ba30a6b

log level

32b578a

Merge branch 'benchmark' of github.com:rudy2steiner/confluo into benc…

5ddc0d8

…hmark merge

timed producer and consumer benchmark

19c8ffa

fix no more message

13f8833

remove old properties

5bbccc6

pub/sub

7fc723f

readme through

153b825

massive test hint

084d227

add annotation and clean

4607f41

properties copy into jar

c95f9ec

anuragkh changed the title ~~Pub sub system benchmark~~ Java pub-sub Implementation Jan 28, 2019

anuragkh requested changes Jan 28, 2019

View reviewed changes

anuragkh requested a review from ujvl January 28, 2019 19:37

anuragkh reviewed Jan 28, 2019

View reviewed changes

javaclient/src/main/java/confluo/streaming/ConfluoConsumer.java Show resolved Hide resolved

anuragkh requested changes Jan 28, 2019

View reviewed changes

rudy2steiner added 5 commits February 11, 2019 23:32

confluo producer and consumer implement closeable interface

7ce8792

code format by google-java-format

400dc7c

ConfluoProducer code format

3e50da8

RecordBatchBuilder test

971fb85

ConfluoConsumer comments

9b900bb

messageListener implement Consumer function

ad5733a

anuragkh requested changes Feb 19, 2019

View reviewed changes

haal reviewed Feb 20, 2019

View reviewed changes

ConfluoProducer/Consumer pub-sub

7f9e0cb

remove useless test

ae73736


		public ConfluoConsumer(Properties properties) throws TException {

		host = properties.getProperty(ConsumerConfig.BOOTSTRAP_ADDRESS, ConsumerConfig.BOOTSTRAP_ADDRESS_DEFAULT);

Java pub-sub Implementation #135

Are you sure you want to change the base?

Java pub-sub Implementation #135

Conversation

rudy2steiner commented Jan 27, 2019 • edited

What changes were proposed in this pull request?

How was this patch tested?

AmplabJenkins commented Jan 27, 2019

anuragkh commented Jan 27, 2019

AmplabJenkins commented Jan 27, 2019

rudy2steiner commented Jan 28, 2019

anuragkh commented Jan 28, 2019

AmplabJenkins commented Jan 28, 2019

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

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

rudy2steiner commented Jan 29, 2019

anuragkh commented Jan 29, 2019

rudy2steiner commented Feb 13, 2019

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

anuragkh commented Feb 19, 2019

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

rudy2steiner commented Feb 21, 2019

anuragkh commented Mar 7, 2019

rudy2steiner commented Mar 8, 2019

anuragkh commented Mar 12, 2019

rudy2steiner commented Mar 14, 2019

AmplabJenkins commented Mar 16, 2021

rudy2steiner commented Jan 27, 2019 •

edited