maedhroz commented on code in PR #23:
URL: https://github.com/apache/cassandra-harry/pull/23#discussion_r1412429423
##########
harry-core/src/harry/model/sut/injvm/InJVMTokenAwareVisitExecutor.java:
##########
@@ -72,44 +78,43 @@ public InJVMTokenAwareVisitExecutor(Run run,
}
@Override
- protected void executeAsyncWithRetries(long lts, long pd,
CompletableFuture<Object[][]> future, CompiledStatement statement)
- {
- executeAsyncWithRetries(lts, pd, future, statement, 0);
- }
-
- private void executeAsyncWithRetries(long lts, long pd,
CompletableFuture<Object[][]> future, CompiledStatement statement, int retries)
+ protected Object[][] executeWithRetries(long lts, long pd,
CompiledStatement statement)
{
if (sut.isShutdown())
throw new IllegalStateException("System under test is shut down");
- if (retries > this.MAX_RETRIES)
- throw new IllegalStateException(String.format("Can not execute
statement %s after %d retries", statement, retries));
+ int retries = 0;
- Object[] pk = schema.inflatePartitionKey(pd);
+ Object[] pk = schema.inflatePartitionKey(pd);
List<TokenPlacementModel.Node> replicas =
getRing().replicasFor(TokenUtil.token(ByteUtils.compose(ByteUtils.objectsToBytes(pk))));
-
- TokenPlacementModel.Node replica = replicas.get((int) (lts %
replicas.size()));
- if (cl == SystemUnderTest.ConsistencyLevel.NODE_LOCAL)
- {
- future.complete(executeNodeLocal(statement.cql(), replica,
statement.bindings()));
- }
- else
+ while (retries++ < MAX_RETRIES)
{
- CompletableFuture.supplyAsync(() -> sut.cluster
- .stream()
- .filter((n) ->
n.config().broadcastAddress().toString().contains(replica.id))
- .findFirst()
- .get()
- .coordinator()
- .execute(statement.cql(),
InJvmSut.toApiCl(cl), statement.bindings()), executor)
- .whenComplete((res, t) ->
- {
- if (t != null)
- executor.schedule(() ->
executeAsyncWithRetries(lts, pd, future, statement, retries + 1), 1,
TimeUnit.SECONDS);
- else
- future.complete(res);
- });
+ try
+ {
+ TokenPlacementModel.Node replica = replicas.get((int) (lts %
replicas.size()));
+ if (cl == SystemUnderTest.ConsistencyLevel.NODE_LOCAL)
+ {
+ return executeNodeLocal(statement.cql(), replica,
statement.bindings());
+ }
+ else
+ {
+ return sut.cluster
+ .stream()
+ .filter((n) ->
n.config().broadcastAddress().toString().contains(replica.id))
+ .findFirst()
+ .get()
+ .coordinator()
+ .execute(statement.cql(), InJvmSut.toApiCl(cl),
statement.bindings());
+ }
+ }
+ catch (Throwable t)
+ {
+ int delaySecs = 1;
+ logger.error(String.format("Caught message while trying to
execute: %s. Scheduled to retry in %s seconds", statement, delaySecs), t);
+ Uninterruptibles.sleepUninterruptibly(delaySecs,
TimeUnit.SECONDS);
+ }
}
+ throw new IllegalStateException(String.format("Can not execute
statement %s after %d retries", statement, retries));
Review Comment:
So tl;dr we made this synchronous and added some logging for visibility?
I'll figure out why I suppose as I look at its callers...
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
