hunk.code_window:     # Number of env steps to optimize for before returning
       "timesteps_per_iteration": 100,
       # === Replay buffer ===
       # Size of the replay buffer. Note that if async_updates is set, then
       # each worker will have a replay buffer of this size.
  
hunk.labels: keep replace keep keep keep keep keep
hunk.after_edit:     # Number of env steps to optimize for before returning.
  
hunk.file_path: rllib/agents/sac/sac.py
hunk.type: replace
file: package org.ray.streaming.runtime.core.processor;
  
  import org.ray.streaming.message.Record;
  import org.ray.streaming.operator.TwoInputOperator;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class TwoInputProcessor<T, O> extends StreamProcessor<Record, TwoInputOperator<T, O>> {
    private static final Logger LOGGER = LoggerFactory.getLogger(TwoInputProcessor.class);
  
    private String leftStream;
    private String rightStream;
  
    public TwoInputProcessor(TwoInputOperator<T, O> operator) {
      super(operator);
    }
  
    @Override
    public void process(Record record) {
      try {
        if (record.getStream().equals(leftStream)) {
          this.operator.processElement(record, null);
        } else {
          this.operator.processElement(null, record);
        }
      } catch (Exception e) {
        throw new RuntimeException(e);
      }
    }
  
    @Override
    public void close() {
      this.operator.close();
    }
  
    public String getLeftStream() {
      return leftStream;
    }
  
    public void setLeftStream(String leftStream) {
      this.leftStream = leftStream;
    }
  
    public String getRightStream() {
      return rightStream;
    }
  
    public void setRightStream(String rightStream) {
      this.rightStream = rightStream;
    }
  }
  