ifesdjeen commented on code in PR #23:
URL: https://github.com/apache/cassandra-harry/pull/23#discussion_r1414603226
##########
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:
Let's say this was the simplest way to know that a query _has_ in fact
finished. This is a barebone model, and probably should be improved upon. But
at least it gives you some ground to stand upon.
--
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]
