[
https://issues.apache.org/jira/browse/CURATOR-623?focusedWorklogId=681440&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-681440
]
ASF GitHub Bot logged work on CURATOR-623:
------------------------------------------
Author: ASF GitHub Bot
Created on: 15/Nov/21 12:10
Start Date: 15/Nov/21 12:10
Worklog Time Spent: 10m
Work Description: eolivelli commented on a change in pull request #401:
URL: https://github.com/apache/curator/pull/401#discussion_r749267842
##########
File path:
curator-recipes/src/main/java/org/apache/curator/framework/recipes/queue/ChildrenCache.java
##########
@@ -66,6 +68,19 @@ public void processResult(CuratorFramework client,
CuratorEvent event) throws Ex
}
};
+ private final ConnectionStateListener connectionStateListener = (__,
newState) -> {
+ if ((newState == ConnectionState.CONNECTED) || (newState ==
ConnectionState.RECONNECTED)) {
+ try
+ {
+ sync();
+ }
+ catch ( Exception e )
+ {
+ throw new RuntimeException(e);
Review comment:
will this error be logged somewhere ?
as we cannot perform any recovery
will it make more sense to simply log the exception and do not rethrow 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.
To unsubscribe, e-mail: dev-unsubscr...@curator.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Issue Time Tracking
-------------------
Worklog Id: (was: 681440)
Time Spent: 0.5h (was: 20m)
> DistributedQueue stops filling after long disconnect from cluster
> -----------------------------------------------------------------
>
> Key: CURATOR-623
> URL: https://issues.apache.org/jira/browse/CURATOR-623
> Project: Apache Curator
> Issue Type: Bug
> Reporter: Никита Соколов
> Assignee: Jordan Zimmerman
> Priority: Major
> Time Spent: 0.5h
> Remaining Estimate: 0h
>
> One of our VMs had network down for 12 minutes and after the network was up,
> the queues have stopped being filled by external processes as curator gave up
> on all watchers. Here is a test reproducing the issue:
> {code:java}
> import junit.framework.TestCase;
> import org.apache.curator.ensemble.fixed.FixedEnsembleProvider;
> import org.apache.curator.framework.CuratorFramework;
> import org.apache.curator.framework.CuratorFrameworkFactory;
> import org.apache.curator.framework.recipes.queue.DistributedQueue;
> import org.apache.curator.framework.recipes.queue.QueueBuilder;
> import org.apache.curator.framework.recipes.queue.QueueConsumer;
> import org.apache.curator.framework.recipes.queue.QueueSerializer;
> import org.apache.curator.framework.state.ConnectionState;
> import org.apache.curator.retry.ExponentialBackoffRetry;
> import org.apache.curator.test.TestingCluster;
> import java.util.concurrent.CompletableFuture;
> import java.util.concurrent.TimeUnit;
> import java.util.function.Consumer;
> public class DistributedQueueTest extends TestCase {
> public void test() throws Exception {
> final var done = new CompletableFuture<>();
> try (
> final var testingCluster = started(new TestingCluster(1));
> final var dyingCuratorFramework =
> getCuratorFramework(testingCluster.getConnectString());
> final var dyingQueue = newQueue(dyingCuratorFramework, item -> {
> if (item.equals("0")) {
> done.complete(null);
> }
> })
> ) {
> dyingQueue.start();
>
> testingCluster.killServer(testingCluster.getInstances().iterator().next());
> Thread.sleep(2 * 60_000);
>
> testingCluster.restartServer(testingCluster.getInstances().iterator().next());
> try (
> final var aliveCuratorFramework =
> getCuratorFramework(testingCluster.getConnectString());
> final var aliveQueue = newQueue(aliveCuratorFramework, __ ->
> {})
> ) {
> aliveQueue.start();
> aliveQueue.put("0");
> done.get(1, TimeUnit.MINUTES);
> }
> }
> }
> private static DistributedQueue<String> newQueue(CuratorFramework
> curatorFramework, Consumer<String> consumer) {
> curatorFramework.start();
> return QueueBuilder.builder(
> curatorFramework,
> new QueueConsumer<String>() {
> @Override
> public void consumeMessage(String o) {
> consumer.accept(o);
> }
> @Override
> public void stateChanged(CuratorFramework curatorFramework,
> ConnectionState connectionState) {
> }
> },
> new QueueSerializer<>() {
> @Override
> public byte[] serialize(String item) {
> return item.getBytes();
> }
> @Override
> public String deserialize(byte[] bytes) {
> return new String(bytes);
> }
> },
> "/MyChildrenCacheTest/queue"
> ).buildQueue();
> }
> private static TestingCluster started(TestingCluster testingCluster)
> throws Exception {
> try {
> testingCluster.start();
> return testingCluster;
> } catch (Throwable throwable) {
> try (testingCluster) {
> throw throwable;
> }
> }
> }
> private static CuratorFramework getCuratorFramework(String connectString)
> {
> return CuratorFrameworkFactory.builder()
> .ensembleProvider(new FixedEnsembleProvider(connectString, true))
> .retryPolicy(new ExponentialBackoffRetry(1000, 3))
> .build();
> }
> } {code}
>
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
