refresh connection for pipeline when getting JedisMovedDataException

[stillerrr]

pipelinedResponses should be clear in MultiNodePipelineBase.java and avoid this pipeline object reporting ENP when executing close() and reuse like #3697

MultiNodePipelineBase can be reused when calling close(), there is no switch to avoid this way to use MultiNodePipelineBase

[codecov-commenter]

Codecov Report

All modified and coverable lines are covered by tests ✅

Comparison is base (64b5aac) 75.64% compared to head (09a1865) 75.65%.

❗ Current head 09a1865 differs from pull request most recent head 4404bbd. Consider uploading reports for the commit 4404bbd to get more accurate results

❗ Your organization needs to install the Codecov GitHub app to enable full functionality.

Additional details and impacted files

@@            Coverage Diff            @@
##             master    #3699   +/-   ##
=========================================
  Coverage     75.64%   75.65%           
- Complexity     4914     4915    +1     
=========================================
  Files           297      297           
  Lines         14956    14957    +1     
  Branches       1124     1124           
=========================================
+ Hits          11314    11316    +2     
  Misses         3149     3149           
+ Partials        493      492    -1     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

[sazzad16]

@stillerrr What about inner Queues in pipelinedResponses?

[stillerrr]

@stillerrr What about inner Queues in pipelinedResponses?

I think it would be okay, the queue would be empty after executing sync(), but pipelinedResponses also has the empty queue

once the pipeline object is reused, there is no connection in connnections map, but has queue in pipelinedResponses map

[stillerrr]

@sazzad16 add method to refresh connection for pipeline when getting JedisMovedDataException

[stillerrr]

@sazzad16 PTAL

[ggivo]

@stillerrr

MultiNodePipelineBase can be reused when calling close(), there is no switch to avoid this way to use MultiNodePipelineBase

Hey, could you provide an example code demonstrating the expected usage?
Do you mean that the expectation is to reuse MultiNodePipelineBase after invoking close() on it?
Adding a unit test to the PR can also help with what we are trying to resolve.

I agree there is currently no switch (and NPE is not good outcome), but for me, using MultiNodePipelineBase after close() is invoked is rather an abuse of the API and I would rather expect that after a pipeline is closed, subsequent command invocation or sync() should throw an error.

[github-actions]

This pull request is marked stale. It will be closed in 30 days if it is not updated.

[snago]

I want to see #3915 fixed as I would be able to remove a log of ugly workarounds from several different projects where I use Jedis, so I'm jumping in as I don't want this stale-closed...

I'm not OP but in their absense I'll offer my interpretation of their intent:

Do you mean that the expectation is to reuse MultiNodePipelineBase after invoking close() on it?

I interpreted the "can" in "MultiNodePipelineBase can be reused when calling close()" as "could" rather than "should". I.e. since there's no guards elsewhere in the class to prevent this, the new functionality in this PR doesn't assume that the class won't be used after close.

[ggivo]

@snago @stillerrr
Thanks for jumping in — I had misinterpreted the request.

If I now understand correctly, the issue isn’t about reusing a ClusterPipeline after it’s been closed. Rather, it’s that the pipeline doesn’t detect slot changes or trigger a slot cache refresh on the ClusterConnectionProvider between sync() calls.

Here’s a concrete example to illustrate the expexcted usage pattern:

try (JedisCluster cluster = new JedisCluster(nodes, DEFAULT_CLIENT_CONFIG);
     ClusterPipeline pipeline = cluster.pipelined()) {

   Response<String> r1 = pipeline.set("key1", "value1");
   Response<String> r2 = pipeline.set("key2", "value2");
   Response<String> r3 = pipeline.set("key3", "value3");

   pipeline.sync();

   assertEquals("OK", r1.get());
   assertEquals("OK", r2.get());
   assertEquals("OK", r3.get());

   Response<String> r4 = pipeline.get("key1");
   Response<String> r5 = pipeline.get("key2");
   Response<String> r6 = pipeline.get("key3");

   pipeline.sync();
   assertThrows(JedisMovedDataException.class, r4::get);
   assertEquals("value1", r4.get());
   assertEquals("value2", r5.get());
   assertEquals("value3", r6.get());
   
   // after pipeline is closed it should throw exception
   pipeline.close();
   
   assertThrows(IllegalStateException.class, ()->pipeline.get("key1"));
 }

Let me know if I’ve now captured the concern accurately

[snago]

Yes, and even worse. If you do all commands in pipelines, all the JedisMovedDataExceptions will get "swallowed" by the pipelines and the JedisCluster will never recover from a cluster change.

[ggivo]

I think in this case, the change makes sense, but we are missing tests to make sure the internal queues are handled properly after replacing the initial connections.

[snago]

If you want to I could try to write some tests.
I would probably have to create a new PR though.

[dengliming]

132qianming

[ggivo]

@snago

if you want to I could try to write some tests.
I would probably have to create a new PR, though.

The team is a bit stretched currently, and help is appreciated.