Trigger multiple one time activities at certain time

Community Support
1

I need to trigger some activities from a workflow at certain intervals.

I have read the HelloPeriodic.java and I have also see this feature request.

Can someone let me know if below is the correct way to implement it using Activity ? Is there any advantage of implementing it using a Scheduled Child Workflow?

private final List<Promise<Boolean>> promiseList = new ArrayList<>();
private final AtomicBoolean exitRequested = new AtomicBoolean(false);

// Finishes the entire workflow
@SignalMethod
public void exitWorkflow() {
    exitRequested.set(true);
}

// Workflow method
@Override
public void initiateRefund(final Message message) {
	log.info("Begin processing refunds for pnrId: {}", message.getPnrId());
	try {
	  // Create refund promises for each refund info.
	  message
	      .getRefundInfoList()
	      .forEach(
	          refundInfo ->
	              createRefundPromise(
	                  refundInfo,
	                  message.getPnrId()));
	  promiseList.forEach(Promise::get);
	} catch (Exception e) {
	  log.error("Exception occurred during initiateRefund, pnrId : {}", message.getPnrId(), e);
	  raise e;                
	}

	log.info("Done processing refunds for pnrId : {}", message.getPnrId());
}

private void createRefundPromise(
  final RefundInfo refundInfo,
  final String pnrId) {
val refundPromise =
    Async.function(
        () -> {
          val duration =
              Duration.between(currentTime(), refundInfo.getRefundDateTime());
          val shouldAbort = Workflow.await(duration, exitRequested::get);
          if (!shouldAbort) {
            refundActivity.processRefund(refundInfo, pnrId);
            log.info(
                "Completed processing refund, seqId: {}, pnrId: {}",
                refundInfo.getSeqId(),
                pnrId);
          } else {
            log.info(
                "Exit requested for refund, seqId: {}, pnrId: {}",
                refundInfo.getSeqId(),
                pnrId);
          }
          return true;
        });
	promiseList.add(refundPromise);
}



private OffsetDateTime currentTime() {
    return OffsetDateTime.ofInstant(
        Instant.ofEpochMilli(Workflow.currentTimeMillis()), ZoneOffset.UTC);
}
2

What is the max possible number of message in the refund list?
There is a couple of things to look into with this approach, one being max number of pending activities per single workflow execution, dynamic config:

limit.numPendingActivities.error default value 2K
Another is the number of possible threads you are creating and holding on to for possibly long time with this approach, as well as timers with workflow.await(duration, condition).
When starting a large number of async activities, another thing you need to be aware of is the combined input sizes of all activities, as request to schedule them are done by a single RespondWorkflowTaskCompleted request to service. Have seen at times this combined size reach the 4mb grpc message size limit and cause issues.

If you could answer first question as to how large this fanout use case can be (num of message, now and what you think it could become as your use case grows over time) I think would be able to mage better suggestion.
One thing that would probably look into doing is batching these requests, or delayed start feature, see sample here. That way you can have workflow id reuse policy to assure no two workflows can create a refund for sample refund id.

Let me know about the max refunds per exec and we can go from there.

3

It is not expected to grow that big. Max is 10 as of now and may become 20 in future but that is not near.

4

Ok then i think your approach should be fine. Would probably try to simplify it a bit (note could be typos just writing this on top of head):

// could return result of activity if needed just fyi
void processRefund(Duration timeout, RefundInfo refundInfo, int pnrId) {
    val shouldAbort = Workflow.await(duration, exitRequested::get);
    if (!shouldAbort) {
      refundActivity.processRefund(refundInfo, pnrId);
    }
  }

// in main workflow method
List<Promise<Void>> processRefundList = new ArrayList();
message
	      .getRefundInfoList()
	      .forEach(
	          refundInfo ->
                      // if you return result from processRefund this needs to be Async.function
	              Promise<Void> refundPromise = Async.procedure(this::processRefund, refundInfo, messaget.getPnrId);
                   processRefundList.add(refundPromise);
));

    Promise.allOf(processRefundList).get();

    for (Promise<String> promise : processRefundList) {
      if (promise.getFailure() == null) {
        // process results
      } else {
        // process failure
     } 
    }
// ...