How to Keep Exactly One OIC Integration Instance running 24/7

September 11, 2020 | 7 minute read
John Graves
PaaS Cloud Integration Architect
Text Size 100%:

Introduction

Oracle Integration has a great feature to periodically run a flow instance according to a schedule.  This schedule can be something simple, like "run every 10 minutes" or something more complex using a calendar ICS definition.  This works great for scenarios that run periodically, then exit.

However, I've recently been tasked to have an OIC flow monitor and process messages from an Oracle Stream using the new Oracle Stream Adapter.  This would ideally require a flow to be running 24/7 to look for messages and process them.

But OIC is not designed to run a single flow instance for more than a few minutes and if we setup a schedule, there will be gaps in time where there are no instances running.

Use-case

My use-case is for the Streams Adapter, but this will work regardless of what is being done inside a given flow.

So what's the secret?  Use the OIC REST APIs for the scheduling service.

The basic idea is:
- Use the scheduler to start the first instance and define the interval.
- Use some basic logic to determine what schedule window we are currently running in.
- When we are in the next schedule window, force the next schedule to start causing it to be in BLOCKING mode.
- Exist the flow
- The next instance will start within a second or two.

Graphically, it looks something like this:

As you will see, even though this is a scheduled flow, the scheduler should only start the first instance.  The process itself will queue up the next instance once it determines it is inside the next schedule window.  In this example, the window time is 10 minutes, but it doesn't really matter what is set here as long as it is short enough to keep a single instance running without OIC shutting it down.

Create Connector

I won't detail the creation of the OIC REST connector, but it should be pretty easy to setup using your OIC URL in the form of:

https://myhost.integration.ocp.oraclecloud.com:443/ic/api

The Flow

Find Next Schedule Window

Once the connector is setup, the first thing we need to do in the flow is make a call to the scheduler API and get a list of current items.

The REST endpoint will look something like this:

/integration/v1/integrations/MYFLOW%7C01.00.0000/schedules/Schedule_MYFLOW_01_00_0000/futureruns

The results will look something like this:

{
"items": [
{
"id": "48233",
"requestId": 48233,
"runTime": "2020-09-11T00:31:15.589+0000",
"runType": "Submitted Run",
"state": "RUNNING",
"submitterName": "john.graves@oracle.com"
},
{
"id": "48234",
"requestId": 48234,
"runTime": "2020-09-11T00:42:13.000+0000",
"runType": "Auto Scheduled Run",
"state": "WAIT",
"submitterName": "john.graves@oracle.com"
},
....
}
 
 

 

The current instance will always be the first item and in the "RUNNING" state. The next scheduled window will be in the second item. The second item might be in a WAIT, BLOCKED or some other state.

So the first step is to call this API and find the second item to know when the next window will start.

Configure the REST adapter to do a GET operation passing the instance ID, version and schedule ID.

Then map the instance information into the REST service.

The integration Id will be something like this:

concat ($self/nsmpr1:metadata/nsmpr1:integration/nsmpr1:identifier, "%7C", $self/nsmpr1:metadata/nsmpr1:integration/nsmpr1:version )

The %7C is the ASCII code for the pipe symbol "|".

The schedule Id will be something like this:

concat ("Schedule_", $self/nsmpr1:metadata/nsmpr1:integration/nsmpr1:identifier, "_", translate ($self/nsmpr1:metadata/nsmpr1:integration/nsmpr1:version, ".", "_" ) )

There is a "Schedule_" prefix and an underscore "_" between the id and version.   Also, the version has underscores rather than the standard dot notation, so we use the translate function.

Then we can grab the result and put it into a variable for our while loop.

To get this value, we will use an xpath similar to this:

$FindTimestampForNextScheduledRun/nsmpr7:executeResponse/nsmpr1:response-wrapper/nsmpr1:items[2]/nsmpr1:runTime

Note: xpath uses a start index of one, so the second item is [2].

If you want, you could check to make sure that at least two items were returned before doing this assignment, but it won't really matter much to the flow.

The While Loop

Now we can start our while look and check to see if the current timestamp is in the next scheduled window.  If it is, we can exit our loop and manage the starting of the next instance.

The while loop check will look something like this:

fn:current-dateTime() < xsd:dateTime($nextWindowTimestamp)

We have to cast the string variable into a dateTime to make the compare work.

All the real work should be done inside this loop.  In my case, I'm calling an Oracle Stream using the Streams Adapter.  This adapter will wait for messages, up to one minute and then exit.  Inside my ProcessMessage scope, I check to see if any messages were found and send them downstream.

Setup Next Instance

Once this flow has run into the next scheduled time window, we want to force a new instance to start.

Note: The reason for this is if we simply exit and we are past the start time for the next instance, the scheduler will SKIP this instance and wait for the next time window.  That's not what we want.

Note: If we tried to exit this instance "just before" the next scheduled instance, we'd have to know exactly how long our loop will run.  And, if we are wrong, there would be no instance in that scheduled window.  Not good. (I go into more detail about this in the demo video below)

So, we wait until AFTER the next window starts to then force a new instance to start.

Again, this is done using the OIC REST API and adapter.

To force the next instance to run, we POST to the API:

/integration/v1/integrations/MYFLOW|01.00.0000/schedule/jobs

with a payload like:

{ "runAsSchedule": true }

We map the instance id and version the same as before and pass "true()" to the 'runAsSchedule" payload.

Note: It is VERY important to pass this value, otherwise you can have multiple instances running at once.

Making this call will create an instance in the schedule set as BLOCKED.  A BLOCKED item will start immediately after the existing instance completes.  This makes it then safe to exit this flow knowing the scheduler will kick off a new instance.

Now, it should never happen, but if something were to go wrong and an extra instance was setup in the schedule queue, you could get multiple BLOCKED instances.  So, I added a little check to see if we already have a BLOCKED instance before making this REST call.  This makes sure we don't get more than one.

This just makes the same call to get the current schedule list and checks to see if there are any with a state of "BLOCKED".

The condition value is:

count($CheckForExistingScheduledRuns/nsmpr7:executeResponse/nssrcdfl:response-wrapper/nssrcdfl:items[nssrcdfl:state='BLOCKED']) > 0

If there is already a BLOCKED schedule, we can simple exit and be certain a new instance will be started.

And that's it.  You will have one and only one instance of this flow running 24/7!

Demo

Here is a quick demo video showing this process in action.

Sample Code

Here is the code if you want to see the example in detail:

https://github.com/gravesjohnr/oicsamples/blob/master/AlwaysRunningFlowSample/TESTSCHEDULEDSTREAMSGETMESSAGE_01.00.0000.iar

Limits

It is important to note that OIC typically limits the number of scheduled processes that are run simultaneously.   Therefore, this patter should be used sparingly.

 

John Graves

PaaS Cloud Integration Architect

Worked for Oracle for many years through the BEA acquisition.  Built many applications using Oracle technology and currently work with all things iPaaS.

Oracle Chatbot
Disconnected