[jira] [Work logged] (CURATOR-623) DistributedQueue stops filling after long disconnect from cluster

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/CURATOR-623?focusedWorklogId=768278&page=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-768278 ]

ASF GitHub Bot logged work on CURATOR-623:
------------------------------------------

                Author: ASF GitHub Bot
            Created on: 10/May/22 02:15
            Start Date: 10/May/22 02:15
    Worklog Time Spent: 10m 
      Work Description: tisonkun commented on PR #401:
URL: https://github.com/apache/curator/pull/401#issuecomment-1121796373

   @Randgalt I've merge master into this branch and tests passed. Would you like to proceed this patch?




Issue Time Tracking
-------------------

    Worklog Id:     (was: 768278)
    Time Spent: 1h 40m  (was: 1.5h)

> 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: 1h 40m
>  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.7#820007)