You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by "gosonzhang (via GitHub)" <gi...@apache.org> on 2023/06/30 08:19:02 UTC

[GitHub] [inlong] gosonzhang opened a new pull request, #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

gosonzhang opened a new pull request, #8386:
URL: https://github.com/apache/inlong/pull/8386

   
   - Fixes #8385
   
   1. Add takeDispatchedRecord() in BufferQueue class
   2. Replace calling pollDispatchedRecord() with takeDispatchedRecord() in the MessageQueueZoneWorker class


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] gosonzhang commented on a diff in pull request #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

Posted by "gosonzhang (via GitHub)" <gi...@apache.org>.
gosonzhang commented on code in PR #8386:
URL: https://github.com/apache/inlong/pull/8386#discussion_r1247638575


##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/BufferQueue.java:
##########
@@ -61,6 +61,19 @@ public A pollRecord() {
         return record;
     }
 
+    /**
+     * Take record
+     */
+    public A takeRecord() {
+        try {
+            A record = queue.take();

Review Comment:
   It's the take() method specially used, mainly to block until there is data in the queue before proceeding with the next processing



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] luchunliang commented on a diff in pull request #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

Posted by "luchunliang (via GitHub)" <gi...@apache.org>.
luchunliang commented on code in PR #8386:
URL: https://github.com/apache/inlong/pull/8386#discussion_r1247594858


##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/BufferQueue.java:
##########
@@ -61,6 +61,19 @@ public A pollRecord() {
         return record;
     }
 
+    /**
+     * Take record
+     */
+    public A takeRecord() {
+        try {
+            A record = queue.take();

Review Comment:
   take operation will lock current thread.
   Please confirm this feature.



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] gosonzhang merged pull request #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

Posted by "gosonzhang (via GitHub)" <gi...@apache.org>.
gosonzhang merged PR #8386:
URL: https://github.com/apache/inlong/pull/8386


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] gosonzhang commented on a diff in pull request #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

Posted by "gosonzhang (via GitHub)" <gi...@apache.org>.
gosonzhang commented on code in PR #8386:
URL: https://github.com/apache/inlong/pull/8386#discussion_r1247629396


##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/BufferQueue.java:
##########
@@ -61,6 +61,19 @@ public A pollRecord() {
         return record;
     }
 
+    /**
+     * Take record
+     */
+    public A takeRecord() {
+        try {
+            A record = queue.take();
+            this.offerCount.incrementAndGet();

Review Comment:
   done



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org

[GitHub] [inlong] luchunliang commented on a diff in pull request #8386: [INLONG-8385][DataProxy] Add take method in BufferQueue class

Posted by "luchunliang (via GitHub)" <gi...@apache.org>.
luchunliang commented on code in PR #8386:
URL: https://github.com/apache/inlong/pull/8386#discussion_r1247592179


##########
inlong-dataproxy/dataproxy-source/src/main/java/org/apache/inlong/dataproxy/utils/BufferQueue.java:
##########
@@ -61,6 +61,19 @@ public A pollRecord() {
         return record;
     }
 
+    /**
+     * Take record
+     */
+    public A takeRecord() {
+        try {
+            A record = queue.take();
+            this.offerCount.incrementAndGet();

Review Comment:
   It is take operation, need to invoke
   this.pollCount.getAndIncrement();



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org