QRunIO/qqq
2
0
Fork 0
mirror of https://github.com/Kingsrook/qqq.git synced 2025-07-18 13:10:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Make pipeLoop minRecords a parameter; add input to getOverrideRecordPipeCapacity

Browse Source
This commit is contained in:
darin.kelkhoff
2023-03-15 17:00:09 -05:00
parent b6e089a364
commit 0e01372200
4 changed files with 60 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/actions/async/AsyncRecordPipeLoop.java
View File

@ -47,7 +47,8 @@ public class AsyncRecordPipeLoop
private static final int MAX_SLEEP_MS = 1000; private static final int MAX_SLEEP_MS = 1000;
private static final int INIT_SLEEP_MS = 10; private static final int INIT_SLEEP_MS = 10;
private static final int MIN_RECORDS_TO_CONSUME = 10;
private Integer minRecordsToConsume = 10;
@ -83,7 +84,7 @@ public class AsyncRecordPipeLoop
while(jobState.equals(AsyncJobState.RUNNING)) while(jobState.equals(AsyncJobState.RUNNING))
{ {
if(recordPipe.countAvailableRecords() < MIN_RECORDS_TO_CONSUME) if(recordPipe.countAvailableRecords() < minRecordsToConsume)
{ {
/////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////
// if the pipe is too empty, sleep to let the producer work. // // if the pipe is too empty, sleep to let the producer work. //
@ -176,4 +177,35 @@ public class AsyncRecordPipeLoop
return (recordCount); return (recordCount);
} }
/*******************************************************************************
** Getter for minRecordsToConsume
*******************************************************************************/
public Integer getMinRecordsToConsume()
{
return (this.minRecordsToConsume);
}
/*******************************************************************************
** Setter for minRecordsToConsume
*******************************************************************************/
public void setMinRecordsToConsume(Integer minRecordsToConsume)
{
this.minRecordsToConsume = minRecordsToConsume;
}
/*******************************************************************************
** Fluent setter for minRecordsToConsume
*******************************************************************************/
public AsyncRecordPipeLoop withMinRecordsToConsume(Integer minRecordsToConsume)
{
this.minRecordsToConsume = minRecordsToConsume;
return (this);
}
} }

2
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/AbstractLoadStep.java
View File

@ -117,7 +117,7 @@ public abstract class AbstractLoadStep implements BackendStep
** In other words, for a slow loader, setting a lower pipe capacity can help prevent ** In other words, for a slow loader, setting a lower pipe capacity can help prevent
** time-out errors ("Giving up adding record to pipe...") ** time-out errors ("Giving up adding record to pipe...")
*******************************************************************************/ *******************************************************************************/
public Integer getOverrideRecordPipeCapacity() public Integer getOverrideRecordPipeCapacity(RunBackendStepInput runBackendStepInput)
{ {
return (null); return (null);
} }

2
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/AbstractTransformStep.java
View File

@ -106,7 +106,7 @@ public abstract class AbstractTransformStep implements BackendStep, ProcessSumma
** In other words, for a slow transformer, setting a lower pipe capacity can help prevent ** In other words, for a slow transformer, setting a lower pipe capacity can help prevent
** time-out errors ("Giving up adding record to pipe...") ** time-out errors ("Giving up adding record to pipe...")
*******************************************************************************/ *******************************************************************************/
public Integer getOverrideRecordPipeCapacity() public Integer getOverrideRecordPipeCapacity(RunBackendStepInput runBackendStepInput)
{ {
return (null); return (null);
} }

25
qqq-backend-core/src/main/java/com/kingsrook/qqq/backend/core/processes/implementations/etl/streamedwithfrontend/StreamedETLExecuteStep.java
View File

@ -80,20 +80,25 @@ public class StreamedETLExecuteStep extends BaseStreamedETLStep implements Backe
// before it can put more records in. // // before it can put more records in. //
///////////////////////////////////////////////////////////////////////////// /////////////////////////////////////////////////////////////////////////////
RecordPipe recordPipe; RecordPipe recordPipe;
if(loadStep.getOverrideRecordPipeCapacity() != null) Integer overrideRecordPipeCapacity = loadStep.getOverrideRecordPipeCapacity(runBackendStepInput);
if(overrideRecordPipeCapacity != null)
{ {
recordPipe = new RecordPipe(loadStep.getOverrideRecordPipeCapacity()); recordPipe = new RecordPipe(overrideRecordPipeCapacity);
LOG.debug("per " + transformStep.getClass().getName() + ", we are overriding record pipe capacity to: " + loadStep.getOverrideRecordPipeCapacity()); LOG.debug("per " + loadStep.getClass().getName() + ", we are overriding record pipe capacity to: " + overrideRecordPipeCapacity);
} }
else if(transformStep.getOverrideRecordPipeCapacity() != null) else
{ {
recordPipe = new RecordPipe(transformStep.getOverrideRecordPipeCapacity()); overrideRecordPipeCapacity = transformStep.getOverrideRecordPipeCapacity(runBackendStepInput);
LOG.debug("per " + transformStep.getClass().getName() + ", we are overriding record pipe capacity to: " + transformStep.getOverrideRecordPipeCapacity()); if(overrideRecordPipeCapacity != null)
{
recordPipe = new RecordPipe(overrideRecordPipeCapacity);
LOG.debug("per " + transformStep.getClass().getName() + ", we are overriding record pipe capacity to: " + overrideRecordPipeCapacity);
} }
else else
{ {
recordPipe = new RecordPipe(); recordPipe = new RecordPipe();
} }
}
extractStep.setRecordPipe(recordPipe); extractStep.setRecordPipe(recordPipe);
extractStep.preRun(runBackendStepInput, runBackendStepOutput); extractStep.preRun(runBackendStepInput, runBackendStepOutput);
@ -113,7 +118,13 @@ public class StreamedETLExecuteStep extends BaseStreamedETLStep implements Backe
} }
List<QRecord> loadedRecordList = new ArrayList<>(); List<QRecord> loadedRecordList = new ArrayList<>();
int recordCount = new AsyncRecordPipeLoop().run("StreamedETL>Execute>ExtractStep", null, recordPipe, (status) -> AsyncRecordPipeLoop asyncRecordPipeLoop = new AsyncRecordPipeLoop();
if(overrideRecordPipeCapacity != null && overrideRecordPipeCapacity < asyncRecordPipeLoop.getMinRecordsToConsume())
{
asyncRecordPipeLoop.setMinRecordsToConsume(overrideRecordPipeCapacity);
}
int recordCount = asyncRecordPipeLoop.run("StreamedETL>Execute>ExtractStep", null, recordPipe, (status) ->
{ {
extractStep.run(runBackendStepInput, runBackendStepOutput); extractStep.run(runBackendStepInput, runBackendStepOutput);
return (runBackendStepOutput); return (runBackendStepOutput);