GEOMESA-3538 Accumulo - Support 'eventual' consistency scans (#3460)

* New parameter `accumulo.query.consistency`
* Remove compatability checks for pre-1.2.4 schemas
* Rename AccumuloDataStore.connector -> client

Author: elahrvivaz

docs/user/accumulo/commandline.rst

@@ -24,10 +24,10 @@ General Arguments
 Most commands require you to specify the connection to Accumulo. This generally includes the instance name,
 zookeeper hosts, username, and password (or Kerberos keytab file). Specify the instance with ``--instance-name``
 and ``--zookeepers``, and the username and password with ``--user`` and ``--password``. The password argument may be
-omitted in order to avoid plaintext credentials in the bash history and process list - in this case it will be
-prompted case for later. To use Kerberos authentication instead of a password, use ``--keytab`` with a path to a
-Kerberos keytab file containing an entry for the specified user. Since a keytab file allows authentication
-without any further constraints, it should be protected appropriately.
+omitted in order to avoid plaintext credentials in the process list - in this case it will be prompted for. To use
+Kerberos authentication instead of a password, use ``--keytab`` with a path to a Kerberos keytab file containing an
+entry for the specified user. Since a keytab file allows authentication without any further constraints, it should be
+protected appropriately.
 
 Instead of specifying the cluster connection explicitly, an appropriate ``accumulo-client.properties``
 may be added to the classpath. See the
@@ -36,7 +36,8 @@ for information on the necessary configuration keys. Any explicit command-line a
 the configuration file.
 
 The ``--auths`` argument corresponds to the ``AccumuloDataStore`` parameter ``geomesa.security.auths``. See
-:ref:`data_security` for more information.
+:ref:`data_security` for more information. The ``--consistency`` argument corresponds to the  ``AccumuloDataStore`` parameter
+``accumulo.query.consistency``.
 
 Commands
 --------

docs/user/accumulo/usage.rst

@@ -32,20 +32,24 @@ Parameter                              Type    Description
 ``accumulo.query.record-threads``      Integer The number of threads to use for record retrieval
 ``accumulo.write.threads``             Integer The number of threads to use for writing records
 ``geomesa.stats.enable``               Boolean Toggle collection of statistics for newly created feature types
+``accumulo.query.consistency``         String  Specify the scan consistency to use for queries. Must be one of ``immediate`` or
+                                               ``eventual``. See the `Accumulo documentation`_ for details.
 ``accumulo.remote.arrow.enable``       Boolean Process Arrow encoding in Accumulo tablets servers as a distributed call
 ``accumulo.remote.bin.enable``         Boolean Process binary encoding in Accumulo tablets servers as a distributed call
 ``accumulo.remote.density.enable``     Boolean Process heatmap encoding in Accumulo tablets servers as a distributed call
 ``accumulo.remote.stats.enable``       Boolean Process statistical calculations in Accumulo tablets servers as a distributed call
 ``geomesa.partition.scan.parallel``    Boolean For partitioned schemas, execute scans in parallel instead of sequentially
 ====================================== ======= ===================================================================================
 
+.. _Accumulo documentation: https://accumulo.apache.org/docs/2.x/apidocs/org/apache/accumulo/core/client/ScannerBase.ConsistencyLevel.html
+
 .. note::
 
     It is an error to specify both ``accumulo.password`` and ``accumulo.keytab.path``.
 
 Instead of specifying the cluster connection explicitly, an appropriate ``accumulo-client.properties`` may be added
 to the classpath. See the
-`Accumulo documentation <https://accumulo.apache.org/docs/2.x/getting-started/clients#creating-an-accumulo-client>`_
+`Accumulo documentation <https://accumulo.apache.org/docs/2.x/getting-started/clients#creating-an-accumulo-client>`__
 for information on the necessary configuration keys. Any explicit data store parameters will take precedence over
 the configuration file.
 
@@ -66,4 +70,4 @@ that the GeoMesa code is on the classpath:
     org.geotools.api.data.DataStore dataStore =
         org.geotools.api.data.DataStoreFinder.getDataStore(parameters);
 
-More information on using GeoTools can be found in the `GeoTools user guide <https://docs.geotools.org/stable/userguide/>`_.
+More information on using GeoTools can be found in the `GeoTools user guide <https://docs.geotools.org/stable/userguide/>`__.

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/audit/AccumuloAuditReader.scala

@@ -9,6 +9,7 @@
 package org.locationtech.geomesa.accumulo.audit
 
 import org.apache.accumulo.core.client.AccumuloClient
+import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel
 import org.apache.accumulo.core.security.Authorizations
 import org.locationtech.geomesa.accumulo.data.AccumuloDataStore
 import org.locationtech.geomesa.index.audit.AuditReader
@@ -24,19 +25,27 @@ import java.time.ZonedDateTime
  * @param client accumulo client - note: assumed to be shared and not cleaned up on closed
  * @param table table containing audit records
  * @param authProvider auth provider
+ * @param consistency scan consistency level
  */
-class AccumuloAuditReader(client: AccumuloClient, table: String, authProvider: AuthorizationsProvider) extends AuditReader {
+class AccumuloAuditReader(
+    client: AccumuloClient,
+    table: String,
+    authProvider: AuthorizationsProvider,
+    consistency: Option[ConsistencyLevel] = None
+  ) extends AuditReader {
 
   import scala.collection.JavaConverters._
 
-  def this(ds: AccumuloDataStore) = this(ds.connector, ds.config.auditWriter.table, ds.config.authProvider)
+  def this(ds: AccumuloDataStore) =
+    this(ds.client, ds.config.auditWriter.table, ds.config.authProvider, ds.config.queries.consistency)
 
   @volatile
   private var tableExists: Boolean = client.tableOperations().exists(table)
 
   override def getQueryEvents(typeName: String, dates: (ZonedDateTime, ZonedDateTime)): CloseableIterator[QueryEvent] = {
     if (!checkTable) { CloseableIterator.empty } else {
       val scanner = client.createScanner(table, new Authorizations(authProvider.getAuthorizations.asScala.toSeq: _*))
+      consistency.foreach(scanner.setConsistencyLevel)
       AccumuloQueryEventTransform.iterator(scanner, typeName, dates)
     }
   }

geomesa-accumulo/geomesa-accumulo-datastore/src/main/scala/org/locationtech/geomesa/accumulo/data/AccumuloBackedMetadata.scala

@@ -9,16 +9,21 @@
 package org.locationtech.geomesa.accumulo.data
 
 import org.apache.accumulo.core.client.AccumuloClient
+import org.apache.accumulo.core.client.ScannerBase.ConsistencyLevel
 import org.apache.accumulo.core.data.{Mutation, Range, Value}
 import org.apache.accumulo.core.security.Authorizations
 import org.apache.hadoop.io.Text
 import org.locationtech.geomesa.accumulo.util.{GeoMesaBatchWriterConfig, TableManager}
-import org.locationtech.geomesa.index.metadata.{GeoMesaMetadata, KeyValueStoreMetadata, MetadataSerializer}
+import org.locationtech.geomesa.index.metadata.{KeyValueStoreMetadata, MetadataSerializer}
 import org.locationtech.geomesa.utils.collection.CloseableIterator
-import org.locationtech.geomesa.utils.io.{CloseQuietly, WithClose}
+import org.locationtech.geomesa.utils.io.WithClose
 
-class AccumuloBackedMetadata[T](val connector: AccumuloClient, val table: String, val serializer: MetadataSerializer[T])
-    extends KeyValueStoreMetadata[T] {
+class AccumuloBackedMetadata[T](
+    val client: AccumuloClient,
+    val table: String,
+    val serializer: MetadataSerializer[T],
+    consistency: Option[ConsistencyLevel] = None
+  ) extends KeyValueStoreMetadata[T] {
 
   import scala.collection.JavaConverters._
 
@@ -28,15 +33,15 @@ class AccumuloBackedMetadata[T](val connector: AccumuloClient, val table: String
 
   private val empty = new Text()
 
-  override protected def checkIfTableExists: Boolean = connector.tableOperations().exists(table)
+  override protected def checkIfTableExists: Boolean = client.tableOperations().exists(table)
 
-  override protected def createTable(): Unit = new TableManager(connector).ensureTableExists(table)
+  override protected def createTable(): Unit = new TableManager(client).ensureTableExists(table)
 
   override protected def createEmptyBackup(timestamp: String): AccumuloBackedMetadata[T] =
-    new AccumuloBackedMetadata(connector, s"${table}_${timestamp}_bak", serializer)
+    new AccumuloBackedMetadata(client, s"${table}_${timestamp}_bak", serializer, consistency)
 
   override protected def write(rows: Seq[(Array[Byte], Array[Byte])]): Unit = {
-    WithClose(connector.createBatchWriter(table, config)) { writer =>
+    WithClose(client.createBatchWriter(table, config)) { writer =>
       rows.foreach { case (k, v) =>
         val m = new Mutation(k)
         m.put(empty, empty, new Value(v))
@@ -47,15 +52,16 @@ class AccumuloBackedMetadata[T](val connector: AccumuloClient, val table: String
 
   override protected def delete(rows: Seq[Array[Byte]]): Unit = {
     val ranges = rows.map(r => Range.exact(new Text(r))).asJava
-    WithClose(connector.createBatchDeleter(table, Authorizations.EMPTY, 1, config)) { deleter =>
+    WithClose(client.createBatchDeleter(table, Authorizations.EMPTY, 1, config)) { deleter =>
       deleter.setRanges(ranges)
       deleter.delete()
     }
   }
 
   override protected def scanValue(row: Array[Byte]): Option[Array[Byte]] = {
-    WithClose(connector.createScanner(table, Authorizations.EMPTY)) { scanner =>
+    WithClose(client.createScanner(table, Authorizations.EMPTY)) { scanner =>
       scanner.setRange(Range.exact(new Text(row)))
+      consistency.foreach(scanner.setConsistencyLevel)
       val iter = scanner.iterator
       if (iter.hasNext) {
         Some(iter.next.getValue.get)
@@ -68,77 +74,12 @@ class AccumuloBackedMetadata[T](val connector: AccumuloClient, val table: String
   override protected def scanRows(prefix: Option[Array[Byte]]): CloseableIterator[(Array[Byte], Array[Byte])] = {
     // ensure we don't scan any single-row encoded values
     val range = prefix.map(p => Range.prefix(new Text(p))).getOrElse(new Range("", "~"))
-    val scanner = connector.createScanner(table, Authorizations.EMPTY)
+    val scanner = client.createScanner(table, Authorizations.EMPTY)
     scanner.setRange(range)
+    consistency.foreach(scanner.setConsistencyLevel)
     CloseableIterator(scanner.iterator.asScala.map(r => (r.getKey.getRow.copyBytes, r.getValue.get)), scanner.close())
   }
-}
-
-object AccumuloBackedMetadata {
-
-  /**
-    * Old single row metadata kept around for back-compatibility
-    */
-  class SingleRowAccumuloMetadata[T](metadata: AccumuloBackedMetadata[T]) {
-
-    import scala.collection.JavaConverters._
-
-    // if the table doesn't exist, we assume that we don't ever need to check it for old-encoded rows
-    private val tableExists = metadata.connector.tableOperations().exists(metadata.table)
 
-    def getFeatureTypes: Array[String] = {
-      if (!tableExists) { Array.empty } else {
-        val scanner = metadata.connector.createScanner(metadata.table, Authorizations.EMPTY)
-        // restrict to just one cf so we only get 1 hit per feature
-        // use attributes as it's the only thing that's been there through all geomesa versions
-        scanner.fetchColumnFamily(new Text(GeoMesaMetadata.AttributesKey))
-        try {
-          scanner.iterator.asScala.map(e => SingleRowAccumuloMetadata.getTypeName(e.getKey.getRow)).toArray
-        } finally {
-          scanner.close()
-        }
-      }
-    }
-
-    /**
-      * Migrate a table from the old single-row metadata to the new
-      *
-      * @param typeName simple feature type name
-      */
-    def migrate(typeName: String): Unit = {
-      if (tableExists) {
-        val scanner = metadata.connector.createScanner(metadata.table, Authorizations.EMPTY)
-        val writer = metadata.connector.createBatchWriter(metadata.table, GeoMesaBatchWriterConfig())
-        try {
-          scanner.setRange(SingleRowAccumuloMetadata.getRange(typeName))
-          scanner.iterator.asScala.foreach { entry =>
-            val key = entry.getKey.getColumnFamily.toString
-            metadata.insert(typeName, key, metadata.serializer.deserialize(typeName, entry.getValue.get))
-            // delete for the old entry
-            val delete = new Mutation(entry.getKey.getRow)
-            delete.putDelete(entry.getKey.getColumnFamily, entry.getKey.getColumnQualifier)
-            writer.addMutation(delete)
-          }
-        } finally {
-          CloseQuietly(scanner)
-          if (writer != null) {
-            CloseQuietly(writer)
-          }
-        }
-      }
-    }
-  }
-
-  object SingleRowAccumuloMetadata {
-
-    private val MetadataTag = "~METADATA"
-    private val MetadataRowKeyRegex = (MetadataTag + """_(.*)""").r
-
-    def getRange(typeName: String): Range = new Range(s"${MetadataTag}_$typeName")
-
-    def getTypeName(row: Text): String = {
-      val MetadataRowKeyRegex(typeName) = row.toString
-      typeName
-    }
-  }
+  @deprecated("Use `client`")
+  def connector: AccumuloClient = client
 }