Create workunitstream for CompactionSource #1826
Conversation
yukuai518
changed the title
| if (last == null) | ||
| log.debug ("Waiting for producer to complete..."); | ||
| } else { | ||
| if (workUnits.size() > 0) { |
There was a problem hiding this comment.
it seems you should add last = this.workUnits.poll() here?
| */ | ||
| public boolean hasNext () { | ||
| try { | ||
| while (true) { |
There was a problem hiding this comment.
Would this work?
while(true) {
if (last != null) return true;
if (this.unProcessed <= 0 && this.workUnits.isEmpty()) return false;
this.last = this.workUnits.poll(1, TimeUnit.SECONDS);
}| last = null; | ||
| log.debug ("next() pops out a workunit"); | ||
| return tmp; | ||
| } |
There was a problem hiding this comment.
you should probably have else { throw new IllegalStateException()} to catch the case where hasNext() returns true but last doesn't get populated.
|
|
||
| List<Dataset> datasets = finder.findDatasets(); | ||
| CompactionDatasetIterator iterator = new CompactionDatasetIterator(datasets, verifiers); | ||
| service.submit(iterator.getDatasetProcessor()); |
There was a problem hiding this comment.
You probably want multiple data processors right? Note that will also require to transform the List into something like a Spliterator (although that is only available on java 8)
|
|
||
| @Override | ||
| public List<WorkUnit> getWorkunits(SourceState state) { |
There was a problem hiding this comment.
You should remove getWorkUnits(SourceState), it is better to only have to maintain one functionally equivalent method.
|
@yukuai518 what is the status of this PR? |
|
Issue: https://issues.apache.org/jira/browse/GOBBLIN-38 Please update your PR title with following prefix: [GOBBLIN-38] |
No description provided.