Page Comparison

public interface PartitionConsumerConfiguration {

  // defines how/where the state is persisted and read from
  public byte[] readState();
  public void persistState(byte[] state);

  // defines a PartitionFilter, defining which partitions to entirely omit from the querying/processing
  public PartitionFilter getPartitionFilter();

  // defines an upper bound on how many partitions can be in the working set at any given time/**
 * Configuration parameters to be used by a {@link PartitionConsumer}.
 */
public class ConsumerConfiguration {

  ...

  /**
   * @return A predicate to be applied on {@link PartitionDetail}s to determine which partitions to include in the
   *         partition consumption.
   */
  public Predicate<PartitionDetail> getPartitionPredicate() {
    return partitionPredicate;
  }

  /**
   * @return An upper bound on the size of the working set of partitions that get serialized as part of the consumer's
   *         state.
   */
  public int getMaxWorkingSetSize(); {
  // defines anreturn uppermaxWorkingSetSize;
bound on how}
many
partitions can be/**
processed by a run* ofDefines aan workflowexpiration instancetimeout, in seconds, publicof int getProcessingLimit();IN_PROGRESS partitions
   //* called@return when processingnumber of theseconds retrievedthat partitionsa ispartition successfulcan be in publicprogress void onSuccess();
 
  // called when there is an error during processing of partitions
  public void onFailure();
}

before it is deemed failed.
   */
  public long getTimeout() {
    return timeout;
  }

public class PartitionConsumer {

  public Iterator<Partition> getPartitions(PartitionedFileSet pfs, PartitionConsumerConfiguration, partitionConsumerConfiguration) {
    PartitionConsumerState state = partitionConsumerConfiguration.readState();
    // get new partitions in the dataset that match a partition PartitionFilter
    Iterator<Partition> partitions =
      pfsDataset.consumePartitions(state.getPointer(), limit = getMaxWorkingSetSize() - state.getSize(), getPartitionFilter());
    // add these new partitions into the working set.
    state.addPartitions(partitions);

    // get a number of partitions from the working set, mark them IN_PROGRESS, and return them.
    Iterator<Partition> partitionsToProcess = state.getPartitions(getProcessingLimit());
    partitionConsumerConfiguration.persistState(state);
    // need to commit this transaction now, so that other instances of the processing entity see these partitions as IN_PROGRESS.
    return partitionsToProcess;
  }
} 

/**
 * Incrementally consumes new/unprocessed {@link Partition}s of a {@link PartitionedFileSet}.
 */
public interface PartitionConsumer {

  /**
   * @return list of {@link PartitionDetail}s that have not yet processed.
   */
  List<PartitionDetail> consumePartitions();

  /**
   * @param limit upper limit on number of partitions to consume
   * @return list of {@link PartitionDetail}s that have not yet processed.
   */
  List<PartitionDetail> consumePartitions(int limit);

  /**
   * This method must be called on any partitions returned by the {@code #consumePartitions} method.
   *
   * @param partitions list of partitions to mark as either succeeded or failed processing
   * @param succeeded whether or not processing of the specified partitions was successful
   */
  void onFinish(List<? extends Partition> partitions, boolean succeeded);

}

/**
 * Worker which runs one iteration of processing new partitions.
 * Note: As we see frequent patterns, we can abstract the common parts of the example usage into CDAP code.
 */
public class CustomWorker extends AbstractWorker {

  @Override
  public void run() {
    PartitionConsumerTransactionalPartitionConsumer partitionConsumer =
      new PartitionConsumerTransactionalPartitionConsumer(getContext();, "lines",
   PartitionConsumerConfiguration pcc = ... // Design of user-defined configuration - TBD     PartitionConsumerResult partitionsToProcess;      getContext().execute(new TxRunnable() {       @Override       public voidnew run(DatasetContext context) throws Exception {
        PartitionedFileSet pfs = context.getDataset("customPFS"); KVTableStatePersistor("consumingState", "state.key"));
    // request new partitions (with an upper limit of 10)
    final List<PartitionDetail> partitions  partitionsToProcess = partitionConsumer.getPartitions(pfs, pccconsumePartitions(10);
 
    } 
   }); 
    boolean success = true;
    try {
      // process partitionsToProcess
    } catch (Throwable t) {
      success = false;
      // log the error
    }
 
    getContext().execute(new TxRunnable() {
      @Override
      public void run(DatasetContext context) throws Exception {
        if (partitionConsumer.onFinish(partitionsToProcess, success);
 {     }
    });
// this will, by default, remove them from the working set because they have successfully been processed
          pcc.onSuccess(partitionsToProcess);
        } else  }
}