You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/12/25 01:22:24 UTC
[GitHub] [pulsar] MarvinCai opened a new pull request #9056: Add streaming dispatcher.
MarvinCai opened a new pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056
Related to #3804
### Motivation
Trying to streamline the dispatcher's read requests to manager ledger instead of micro batch.
### Modifications
Created a StreamingEntryReader that can streamline read request to managed ledger.
Created StreamingDispatcher interface with necessary method to interact with StreamingEntryReader.
Created PersistentStreamingDispatcherSingleActive/MultipleConsumer that make use of StreamingEntryReader to read entries from managed ledger.
Add config to use streaming dispatcher.
### Verifying this change
- [ ] Make sure that the change passes the CI checks.
*(Please pick either of the following options)*
This change added tests and can be verified as follows:
Add unit tests.
### Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API: no
- The schema: no
- The default values of configurations: no
- The wire protocol: no
- The rest endpoints: no
- The admin cli options: no
- Anything that affects deployment: no
### Documentation
- Does this pull request introduce a new feature? yes
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771208486
@sijie Rebased to newer version of master and resolved conflict.
Also I saw original Dispatcher logic has some changes related to Transaction, will have separate pr to update streaming Dispatcher if needed.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-751301282
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r557711929
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java
##########
@@ -0,0 +1,338 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.streamingdispatch;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.WaitingEntryCallBack;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.util.SafeRun;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.client.impl.Backoff;
+
+/**
+ * Entry reader that fulfill read request by streamline the read instead of reading with micro batch.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class StreamingEntryReader implements AsyncCallbacks.ReadEntryCallback, WaitingEntryCallBack {
Review comment:
@MarvinCai Can you write a test case for this 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-760490194
@codelipenghui @hangc0276 Can you review this pull request?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771166309
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai edited a comment on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai edited a comment on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771208486
@sijie Rebased to newer version of master and resolved conflict.
Also I saw original Dispatcher logic has some changes related to Transaction, will have separate pr to update streaming Dispatcher if needed to be compatible with transaction.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r552316032
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1915,6 +1920,21 @@ void notifyCursors() {
}
}
+ void notifyWaitingEntryCallBack() {
Review comment:
updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-770527299
@MarvinCai Can you rebase this pull request?
@codelipenghui Can you review it?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r552316433
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
##########
@@ -667,6 +667,12 @@
)
private boolean preciseDispatcherFlowControl = false;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Whether to use streaming read dispatcher."
+ )
+ private boolean streamingDispatch = false;
Review comment:
updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771116333
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r557765792
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/streamingdispatch/StreamingEntryReader.java
##########
@@ -0,0 +1,338 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.streamingdispatch;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.WaitingEntryCallBack;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.util.SafeRun;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
+import org.apache.pulsar.broker.transaction.buffer.exceptions.TransactionNotSealedException;
+import org.apache.pulsar.client.impl.Backoff;
+
+/**
+ * Entry reader that fulfill read request by streamline the read instead of reading with micro batch.
+ */
+@Slf4j
+@RequiredArgsConstructor
+public class StreamingEntryReader implements AsyncCallbacks.ReadEntryCallback, WaitingEntryCallBack {
Review comment:
There's StreamingEntryReaderTests class, it's folded by github.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771180945
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-756590914
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-763998546
@codelipenghui Can you review this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771221855
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r552316032
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1915,6 +1920,21 @@ void notifyCursors() {
}
}
+ void notifyWaitingEntryCallBack() {
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai edited a comment on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai edited a comment on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771208486
@sijie Rebased to newer version of master and resolved conflict.
Also I saw original Dispatcher logic has some changes related to Transaction, will have separate pr to update streaming Dispatcher if needed to be compatible with transaction.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r549788792
##########
File path: managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java
##########
@@ -1915,6 +1920,21 @@ void notifyCursors() {
}
}
+ void notifyWaitingEntryCallBack() {
Review comment:
```suggestion
void notifyWaitingEntryCallBacks() {
```
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
##########
@@ -667,6 +667,12 @@
)
private boolean preciseDispatcherFlowControl = false;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Whether to use streaming read dispatcher."
+ )
+ private boolean streamingDispatch = false;
Review comment:
Can you mark this feature as a preview feature?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
##########
@@ -0,0 +1,191 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import com.google.common.collect.Lists;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.util.SafeRun;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.streamingdispatch.PendingReadEntryRequest;
+import org.apache.pulsar.broker.service.streamingdispatch.StreamingDispatcher;
+import org.apache.pulsar.broker.service.streamingdispatch.StreamingEntryReader;
+
+/**
+ * A {@link PersistentDispatcherMultipleConsumers} implemented {@link StreamingDispatcher}.
+ * It'll use {@link StreamingEntryReader} to read new entries instead read as micro batch from managed ledger.
+ */
+@Slf4j
+public class PersistentStreamingDispatcherMultipleConsumers extends PersistentDispatcherMultipleConsumers
+ implements StreamingDispatcher {
+
+ private StreamingEntryReader streamingEntryReader = new StreamingEntryReader((ManagedCursorImpl) cursor,
Review comment:
final?
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
##########
@@ -667,6 +667,12 @@
)
private boolean preciseDispatcherFlowControl = false;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Whether to use streaming read dispatcher."
Review comment:
It would be great to add more documentation about 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-756776085
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-756776085
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-771221855
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on a change in pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on a change in pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#discussion_r552316433
##########
File path: pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
##########
@@ -667,6 +667,12 @@
)
private boolean preciseDispatcherFlowControl = false;
+ @FieldContext(
+ category = CATEGORY_SERVER,
+ doc = "Whether to use streaming read dispatcher."
+ )
+ private boolean streamingDispatch = false;
Review comment:
updated the doc.
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStreamingDispatcherMultipleConsumers.java
##########
@@ -0,0 +1,191 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.service.persistent;
+
+import com.google.common.collect.Lists;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.bookkeeper.mledger.Position;
+import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
+import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
+import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.mledger.util.SafeRun;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.streamingdispatch.PendingReadEntryRequest;
+import org.apache.pulsar.broker.service.streamingdispatch.StreamingDispatcher;
+import org.apache.pulsar.broker.service.streamingdispatch.StreamingEntryReader;
+
+/**
+ * A {@link PersistentDispatcherMultipleConsumers} implemented {@link StreamingDispatcher}.
+ * It'll use {@link StreamingEntryReader} to read new entries instead read as micro batch from managed ledger.
+ */
+@Slf4j
+public class PersistentStreamingDispatcherMultipleConsumers extends PersistentDispatcherMultipleConsumers
+ implements StreamingDispatcher {
+
+ private StreamingEntryReader streamingEntryReader = new StreamingEntryReader((ManagedCursorImpl) cursor,
Review comment:
updated.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] MarvinCai commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
MarvinCai commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-751305454
/pulsarbot run-failure-checks
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056#issuecomment-763998546
@codelipenghui Can you review this?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui merged pull request #9056: Add streaming dispatcher.
Posted by GitBox <gi...@apache.org>.
codelipenghui merged pull request #9056:
URL: https://github.com/apache/pulsar/pull/9056
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org