class JobOperationsImpl extends HasMetadataOperation<Job, JobList, ScalableResource<Job>>
    implements ScalableResource<Job> {

  static final transient Logger LOG = LoggerFactory.getLogger(JobOperationsImpl.class);
  private final PodOperationContext podControllerOperationContext;

  public JobOperationsImpl(Client client) {
    this(new PodOperationContext(), HasMetadataOperationsImpl.defaultContext(client));
  }

  public JobOperationsImpl(PodOperationContext context, OperationContext superContext) {
    super(superContext.withApiGroupName("batch")
        .withApiGroupVersion("v1")
        .withPlural("jobs"), Job.class, JobList.class);
    this.podControllerOperationContext = context;
  }

  @Override
  public JobOperationsImpl newInstance(OperationContext context) {
    return new JobOperationsImpl(podControllerOperationContext, context);
  }

  @Override
  public Job scale(int count) {
    Job res = accept(b -> b.getSpec().setParallelism(count));
    if (context.getTimeout() > 0) {
      waitUntilJobIsScaled();
      res = get();
    }
    return res;
  }

  /**
   * Lets wait until there are enough Ready pods of the given Job
   */
  private void waitUntilJobIsScaled() {<FILL_FUNCTION_BODY>}

  @Override
  public String getLog() {
    return getLog(podControllerOperationContext.isPrettyOutput());
  }

  @Override
  public String getLog(boolean isPretty) {
    StringBuilder stringBuilder = new StringBuilder();
    List<PodResource> podOperationList = doGetLog();
    for (PodResource podOperation : podOperationList) {
      stringBuilder.append(podOperation.getLog(isPretty));
    }
    return stringBuilder.toString();
  }

  private List<PodResource> doGetLog() {
    Job job = requireFromServer();

    return PodOperationUtil.getPodOperationsForController(context, podControllerOperationContext,
        job.getMetadata().getUid(),
        getJobPodLabels(job));
  }

  /**
   * Returns an unclosed Reader. It's the caller responsibility to close it.
   *
   * @return Reader
   */
  @Override
  public Reader getLogReader() {
    return PodOperationUtil.getLogReader(doGetLog());
  }

  /**
   * Returns an unclosed InputStream. It's the caller responsibility to close it.
   *
   * @return Reader
   */
  @Override
  public InputStream getLogInputStream() {
    return PodOperationUtil.getLogInputStream(doGetLog());
  }

  @Override
  public LogWatch watchLog() {
    return watchLog(null);
  }

  @Override
  public LogWatch watchLog(OutputStream out) {
    return PodOperationUtil.watchLog(doGetLog(), out);
  }

  @Override
  public Loggable withLogWaitTimeout(Integer logWaitTimeout) {
    return withReadyWaitTimeout(logWaitTimeout);
  }

  @Override
  public Loggable withReadyWaitTimeout(Integer timeout) {
    return new JobOperationsImpl(podControllerOperationContext.withReadyWaitTimeout(timeout), context);
  }

  @Override
  protected Job modifyItemForReplaceOrPatch(Supplier<Job> current, Job job) {
    Job jobFromServer = current.get();
    if (job.getSpec().getSelector() == null) {
      job.getSpec().setSelector(jobFromServer.getSpec().getSelector());
    }
    if (job.getSpec().getTemplate().getMetadata() != null) {
      job.getSpec().getTemplate().getMetadata().setLabels(jobFromServer.getSpec().getTemplate().getMetadata().getLabels());
    } else {
      job.getSpec().getTemplate().setMetadata(jobFromServer.getSpec().getTemplate().getMetadata());
    }
    return job;
  }

  static Map<String, String> getJobPodLabels(Job job) {
    Map<String, String> labels = new HashMap<>();
    if (job != null && job.getMetadata() != null && job.getMetadata().getUid() != null) {
      labels.put("controller-uid", job.getMetadata().getUid());
    }
    return labels;
  }

  @Override
  public Loggable inContainer(String id) {
    return new JobOperationsImpl(podControllerOperationContext.withContainerId(id), context);
  }

  @Override
  public TimeTailPrettyLoggable limitBytes(int limitBytes) {
    return new JobOperationsImpl(podControllerOperationContext.withLimitBytes(limitBytes), context);
  }

  @Override
  public TimeTailPrettyLoggable terminated() {
    return new JobOperationsImpl(podControllerOperationContext.withTerminatedStatus(true), context);
  }

  @Override
  public Loggable withPrettyOutput() {
    return new JobOperationsImpl(podControllerOperationContext.withPrettyOutput(true), context);
  }

  @Override
  public PrettyLoggable tailingLines(int lines) {
    return new JobOperationsImpl(podControllerOperationContext.withTailingLines(lines), context);
  }

  @Override
  public TailPrettyLoggable sinceTime(String timestamp) {
    return new JobOperationsImpl(podControllerOperationContext.withSinceTimestamp(timestamp), context);
  }

  @Override
  public TailPrettyLoggable sinceSeconds(int seconds) {
    return new JobOperationsImpl(podControllerOperationContext.withSinceSeconds(seconds), context);
  }

  @Override
  public BytesLimitTerminateTimeTailPrettyLoggable usingTimestamps() {
    return new JobOperationsImpl(podControllerOperationContext.withTimestamps(true), context);
  }
}

    final AtomicReference<Job> atomicJob = new AtomicReference<>();

    waitUntilCondition(job -> {
      atomicJob.set(job);
      Integer activeJobs = job.getStatus().getActive();
      if (activeJobs == null) {
        activeJobs = 0;
      }
      if (Objects.equals(job.getSpec().getParallelism(), activeJobs)) {
        return true;
      }
      LOG.debug("Only {}/{} pods scheduled for Job: {} in namespace: {} so waiting...",
          job.getStatus().getActive(), job.getSpec().getParallelism(), job.getMetadata().getName(), namespace);
      return false;
    }, context.getTimeout(), context.getTimeoutUnit());
  
<methods>public void <init>(io.fabric8.kubernetes.client.dsl.internal.OperationContext, Class<Job>, Class<JobList>) ,public Job accept(Consumer<Job>) ,public Job edit(UnaryOperator<Job>) ,public transient Job edit(Visitor[]) ,public Job editStatus(UnaryOperator<Job>) ,public HasMetadataOperation<Job,JobList,ScalableResource<Job>> newInstance(io.fabric8.kubernetes.client.dsl.internal.OperationContext) ,public Job patch() ,public Job patch(io.fabric8.kubernetes.client.dsl.base.PatchContext) ,public Job patch(io.fabric8.kubernetes.client.dsl.base.PatchContext, Job) ,public Job patch(io.fabric8.kubernetes.client.dsl.base.PatchContext, java.lang.String) ,public Job patchStatus() ,public Job patchStatus(Job) ,public Job replace() ,public Job replaceStatus() ,public Job scale(int) ,public Job scale(int, boolean) ,public Scale scale(Scale) ,public Job update() ,public Job updateStatus() <variables>public static final long DEFAULT_GRACE_PERIOD_IN_SECONDS,public static final io.fabric8.kubernetes.api.model.DeletionPropagation DEFAULT_PROPAGATION_POLICY,private static final Logger LOGGER,private static final java.lang.String PATCH_OPERATION,private static final java.lang.String REPLACE_OPERATION,private static final java.lang.String UPDATE_OPERATION