<commit_before>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.accumulo.core.client.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.apache.accumulo.cloudtrace.instrument.TraceRunnable;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.Column;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.data.thrift.InitialMultiScan;
import org.apache.accumulo.core.data.thrift.MultiScanResult;
import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TKeyValue;
import org.apache.accumulo.core.data.thrift.TRange;
import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.thrift.AuthInfo;
import org.apache.accumulo.core.security.thrift.ThriftSecurityException;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.OpTimer;
import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;


public class TabletServerBatchReaderIterator implements Iterator<Entry<Key,Value>> {
  
  private static final Logger log = Logger.getLogger(TabletServerBatchReaderIterator.class);
  
  private final Instance instance;
  private final AuthInfo credentials;
  private final String table;
  private Authorizations authorizations = Constants.NO_AUTHS;
  private final int numThreads;
  private final ExecutorService queryThreadPool;
  private final ScannerOptions options;
  
  private ArrayBlockingQueue<Entry<Key,Value>> resultsQueue = new ArrayBlockingQueue<Entry<Key,Value>>(1000);
  private Entry<Key,Value> nextEntry = null;
  private Object nextLock = new Object();
  
  private long failSleepTime = 100;
  
  private volatile Throwable fatalException = null;
  
  public interface ResultReceiver {
    void receive(Key key, Value value);
  }
  
  private static class MyEntry implements Entry<Key,Value> {
    
    private Key key;
    private Value value;
    
    MyEntry(Key key, Value value) {
      this.key = key;
      this.value = value;
    }
    
    @Override
    public Key getKey() {
      return key;
    }
    
    @Override
    public Value getValue() {
      return value;
    }
    
    @Override
    public Value setValue(Value value) {
      throw new UnsupportedOperationException();
    }
    
  }
  
  public TabletServerBatchReaderIterator(Instance instance, AuthInfo credentials, String table, Authorizations authorizations, ArrayList<Range> ranges,
      int numThreads, ExecutorService queryThreadPool, ScannerOptions scannerOptions) {
    
    this.instance = instance;
    this.credentials = credentials;
    this.table = table;
    this.authorizations = authorizations;
    this.numThreads = numThreads;
    this.queryThreadPool = queryThreadPool;
    this.options = new ScannerOptions(scannerOptions);
    
    if (options.fetchedColumns.size() > 0) {
      ArrayList<Range> ranges2 = new ArrayList<Range>(ranges.size());
      for (Range range : ranges) {
        ranges2.add(range.bound(options.fetchedColumns.first(), options.fetchedColumns.last()));
      }
      
      ranges = ranges2;
    }
    
    ResultReceiver rr = new ResultReceiver() {
      
      @Override
      public void receive(Key key, Value value) {
        try {
          resultsQueue.put(new MyEntry(key, value));
        } catch (InterruptedException e) {
          if (TabletServerBatchReaderIterator.this.queryThreadPool.isShutdown())
            log.debug("Failed to add Batch Scan result for key " + key, e);
          else
            log.warn("Failed to add Batch Scan result for key " + key, e);
          fatalException = e;
          throw new RuntimeException(e);
          
        }
      }
      
    };
    
    try {
      lookup(ranges, rr);
    } catch (RuntimeException re) {
      throw re;
    } catch (Exception e) {
      throw new RuntimeException("Failed to create iterator", e);
    }
  }
  
  @Override
  public boolean hasNext() {
    synchronized (nextLock) {
      // check if one was cached
      if (nextEntry != null)
        return nextEntry.getKey() != null && nextEntry.getValue() != null;
      
      // don't have one cached, try to cache one and return success
      try {
        while (nextEntry == null && fatalException == null)
          nextEntry = resultsQueue.poll(1, TimeUnit.SECONDS);
        
        if (fatalException != null)
          if (fatalException instanceof RuntimeException)
            throw (RuntimeException) fatalException;
          else
            throw new RuntimeException(fatalException);
        
        return nextEntry.getKey() != null && nextEntry.getValue() != null;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
  
  @Override
  public Entry<Key,Value> next() {
    Entry<Key,Value> current = null;
    
    // if there's one waiting, or hasNext() can get one, return it
    synchronized (nextLock) {<commit_msg>ACCUMULO-487 Made batch scanner iterator fail when batch scanner closed<commit_after>/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.accumulo.core.client.impl;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

import org.apache.accumulo.cloudtrace.instrument.TraceRunnable;
import org.apache.accumulo.core.Constants;
import org.apache.accumulo.core.client.AccumuloException;
import org.apache.accumulo.core.client.AccumuloSecurityException;
import org.apache.accumulo.core.client.Instance;
import org.apache.accumulo.core.client.TableDeletedException;
import org.apache.accumulo.core.client.TableNotFoundException;
import org.apache.accumulo.core.client.TableOfflineException;
import org.apache.accumulo.core.conf.AccumuloConfiguration;
import org.apache.accumulo.core.data.Column;
import org.apache.accumulo.core.data.Key;
import org.apache.accumulo.core.data.KeyExtent;
import org.apache.accumulo.core.data.Range;
import org.apache.accumulo.core.data.Value;
import org.apache.accumulo.core.data.thrift.InitialMultiScan;
import org.apache.accumulo.core.data.thrift.MultiScanResult;
import org.apache.accumulo.core.data.thrift.TKeyExtent;
import org.apache.accumulo.core.data.thrift.TKeyValue;
import org.apache.accumulo.core.data.thrift.TRange;
import org.apache.accumulo.core.master.state.tables.TableState;
import org.apache.accumulo.core.security.Authorizations;
import org.apache.accumulo.core.security.thrift.AuthInfo;
import org.apache.accumulo.core.security.thrift.ThriftSecurityException;
import org.apache.accumulo.core.tabletserver.thrift.NoSuchScanIDException;
import org.apache.accumulo.core.tabletserver.thrift.TabletClientService;
import org.apache.accumulo.core.util.ByteBufferUtil;
import org.apache.accumulo.core.util.OpTimer;
import org.apache.accumulo.core.util.ThriftUtil;
import org.apache.accumulo.core.util.UtilWaitThread;
import org.apache.hadoop.io.Text;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.thrift.TApplicationException;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;


public class TabletServerBatchReaderIterator implements Iterator<Entry<Key,Value>> {
  
  private static final Logger log = Logger.getLogger(TabletServerBatchReaderIterator.class);
  
  private final Instance instance;
  private final AuthInfo credentials;
  private final String table;
  private Authorizations authorizations = Constants.NO_AUTHS;
  private final int numThreads;
  private final ExecutorService queryThreadPool;
  private final ScannerOptions options;
  
  private ArrayBlockingQueue<Entry<Key,Value>> resultsQueue = new ArrayBlockingQueue<Entry<Key,Value>>(1000);
  private Entry<Key,Value> nextEntry = null;
  private Object nextLock = new Object();
  
  private long failSleepTime = 100;
  
  private volatile Throwable fatalException = null;
  
  public interface ResultReceiver {
    void receive(Key key, Value value);
  }
  
  private static class MyEntry implements Entry<Key,Value> {
    
    private Key key;
    private Value value;
    
    MyEntry(Key key, Value value) {
      this.key = key;
      this.value = value;
    }
    
    @Override
    public Key getKey() {
      return key;
    }
    
    @Override
    public Value getValue() {
      return value;
    }
    
    @Override
    public Value setValue(Value value) {
      throw new UnsupportedOperationException();
    }
    
  }
  
  public TabletServerBatchReaderIterator(Instance instance, AuthInfo credentials, String table, Authorizations authorizations, ArrayList<Range> ranges,
      int numThreads, ExecutorService queryThreadPool, ScannerOptions scannerOptions) {
    
    this.instance = instance;
    this.credentials = credentials;
    this.table = table;
    this.authorizations = authorizations;
    this.numThreads = numThreads;
    this.queryThreadPool = queryThreadPool;
    this.options = new ScannerOptions(scannerOptions);
    
    if (options.fetchedColumns.size() > 0) {
      ArrayList<Range> ranges2 = new ArrayList<Range>(ranges.size());
      for (Range range : ranges) {
        ranges2.add(range.bound(options.fetchedColumns.first(), options.fetchedColumns.last()));
      }
      
      ranges = ranges2;
    }
    
    ResultReceiver rr = new ResultReceiver() {
      
      @Override
      public void receive(Key key, Value value) {
        try {
          resultsQueue.put(new MyEntry(key, value));
        } catch (InterruptedException e) {
          if (TabletServerBatchReaderIterator.this.queryThreadPool.isShutdown())
            log.debug("Failed to add Batch Scan result for key " + key, e);
          else
            log.warn("Failed to add Batch Scan result for key " + key, e);
          fatalException = e;
          throw new RuntimeException(e);
          
        }
      }
      
    };
    
    try {
      lookup(ranges, rr);
    } catch (RuntimeException re) {
      throw re;
    } catch (Exception e) {
      throw new RuntimeException("Failed to create iterator", e);
    }
  }
  
  @Override
  public boolean hasNext() {
    synchronized (nextLock) {
      // check if one was cached
      if (nextEntry != null)
        return nextEntry.getKey() != null && nextEntry.getValue() != null;
      
      // don't have one cached, try to cache one and return success
      try {
        while (nextEntry == null && fatalException == null && !queryThreadPool.isShutdown())
          nextEntry = resultsQueue.poll(1, TimeUnit.SECONDS);
        
        if (fatalException != null)
          if (fatalException instanceof RuntimeException)
            throw (RuntimeException) fatalException;
          else
            throw new RuntimeException(fatalException);
        
        if (queryThreadPool.isShutdown())
          throw new RuntimeException("scanner closed");

        return nextEntry.getKey() != null && nextEntry.getValue() != null;
      } catch (InterruptedException e) {
        throw new RuntimeException(e);
      }
    }
  }
  
  @Override
  public Entry<Key,Value> next() {
    Entry<Key,Value> current = null;
    
    // if there's one waiting, or hasNext() can get one, return it
    synchronized (nextLock) {
{'fraction_non_alphanumeric': 0.0925571213572308, 'fraction_numerical': 0.0026679436311396907, 'mean_word_length': 3.5741551939924907, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3441896', 'n_tokens_mistral': 4170, 'n_tokens_neox': 4112, 'n_words': 1060}