// 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.

#include <cstddef>
#include <cstdint>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include <boost/optional/optional.hpp>
#include <gflags/gflags_declare.h>
#include <glog/logging.h>
#include <gtest/gtest.h>

#include "kudu/client/client-test-util.h"
#include "kudu/client/client.h"
#include "kudu/client/scan_batch.h"
#include "kudu/client/scan_configuration.h"
#include "kudu/client/scan_predicate.h"
#include "kudu/client/scanner-internal.h"
#include "kudu/client/schema.h"
#include "kudu/client/shared_ptr.h" // IWYU pragma: keep
#include "kudu/client/value.h"
#include "kudu/client/write_op.h"
#include "kudu/clock/clock.h"
#include "kudu/clock/hybrid_clock.h"
#include "kudu/clock/mock_ntp.h"
#include "kudu/clock/time_service.h"
#include "kudu/common/partial_row.h"
#include "kudu/gutil/casts.h"
#include "kudu/gutil/ref_counted.h"
#include "kudu/gutil/stl_util.h"
#include "kudu/gutil/stringprintf.h"
#include "kudu/gutil/strings/substitute.h"
#include "kudu/integration-tests/internal_mini_cluster-itest-base.h"
#include "kudu/master/catalog_manager.h"
#include "kudu/master/master.h"
#include "kudu/master/master.pb.h"
#include "kudu/master/mini_master.h"
#include "kudu/mini-cluster/internal_mini_cluster.h"
#include "kudu/tablet/tablet_replica.h"
#include "kudu/tserver/mini_tablet_server.h"
#include "kudu/tserver/tablet_server.h"
#include "kudu/tserver/ts_tablet_manager.h"
#include "kudu/util/monotime.h"
#include "kudu/util/random.h"
#include "kudu/util/slice.h"
#include "kudu/util/status.h"
#include "kudu/util/test_macros.h"
#include "kudu/util/test_util.h"

DECLARE_int32(heartbeat_interval_ms);
DECLARE_int32(max_clock_sync_error_usec);
DECLARE_int32(scanner_gc_check_interval_us);
DECLARE_string(time_source);

using boost::none;
using kudu::client::ScanConfiguration;
using kudu::client::sp::shared_ptr;
using kudu::master::CatalogManager;
using kudu::master::GetTableLocationsRequestPB;
using kudu::master::GetTableLocationsResponsePB;
using kudu::clock::HybridClock;
using kudu::tablet::TabletReplica;
using kudu::tserver::MiniTabletServer;
using kudu::tserver::TabletServer;
using std::string;
using std::thread;
using std::vector;
using std::unique_ptr;
using strings::Substitute;

namespace kudu {
namespace client {


class ConsistencyITest : public MiniClusterITestBase {
 public:
  ConsistencyITest()
      : num_tablet_servers_(3),
        table_name_("timestamp_propagation_test_table"),
        key_column_name_("key"),
        key_split_value_(8) {

    // Using the mock clock: need to advance the clock for tablet servers.
    FLAGS_time_source = "mock";

    // Reduce the TS<->Master heartbeat interval: this speeds up testing,
    // saving about 700ms per test.
    FLAGS_heartbeat_interval_ms = 10;
    FLAGS_scanner_gc_check_interval_us = 50 * 1000;

    KuduSchemaBuilder b;
    b.AddColumn(key_column_name_)->Type(KuduColumnSchema::INT32)
        ->NotNull()->PrimaryKey();
    b.AddColumn("int_val")->Type(KuduColumnSchema::INT32)->NotNull();
    b.AddColumn("string_val")->Type(KuduColumnSchema::STRING)->Nullable();
    CHECK_OK(b.Build(&schema_));
  }

  virtual void SetUp() override {
    MiniClusterITestBase::SetUp();
    StartCluster(num_tablet_servers_);

    // Since we're using mock NTP rather than the hybrid clock, it's possible
    // that the first timestamp assigned to a tablet message is the initial
    // timestamp (0). For correctness of scans, it is illegal to scan in this
    // state. As such, we bump the clock up front so when we create tablets,
    // they start out with more natural, non-0 values.
    for (int i = 0; i < num_tablet_servers_; i++) {
      cluster_->mini_tablet_server(i)->server()->clock()->Now();
    }
  }

  void ScannerThread(KuduClient::ReplicaSelection selection,
                     int rows_to_insert,
                     int first_row,
                     int scans_to_perform) {
    client::sp::shared_ptr<KuduClient> client;
    CHECK_OK(cluster_->CreateClient(nullptr, &client));

    shared_ptr<KuduTable> table;
    CHECK_OK(client->OpenTable(table_name_, &table));

    size_t row_count;
    for (int i = 0; i < 3; i++) {
      // Insert multiple rows into the tablets.
      InsertTestRows(client.get(), table.get(), rows_to_insert, first_row * i);
      int expected_count = rows_to_insert * (i + 1);
      // Perform a bunch of READ_YOUR_WRITES scans to all the replicas
      // that count the rows. And verify that the count of the rows
      // never go down from what previously observed, to ensure subsequent
      // reads will not "go back in time" regarding writes that other
      // clients have done.
      for (int j = 0; j < scans_to_perform; j++) {
        CHECK_OK(GetRowCount(table.get(), KuduScanner::READ_YOUR_WRITES,
                             selection, 0, &row_count));
        EXPECT_LE(expected_count, row_count);
        expected_count = row_count;
      }
    }
  }

 protected:
  static void UpdateClock(HybridClock* clock, MonoDelta delta) {
    const uint64_t new_time(HybridClock::GetPhysicalValueMicros(clock->Now()) +
                            delta.ToMicroseconds());
    auto* ntp = down_cast<clock::MockNtp*>(clock->time_service());
    ntp->SetMockClockWallTimeForTests(new_time);
  }

  // Creates a table with the specified name and replication factor.
  Status CreateTable(KuduClient* client,
                     const string& table_name,
                     int num_replicas = 1) {
    unique_ptr<KuduPartialRow> split_row(schema_.NewRow());
    RETURN_NOT_OK(split_row->SetInt32(0, key_split_value_));

    unique_ptr<KuduTableCreator> table_creator(client->NewTableCreator());
    RETURN_NOT_OK(table_creator->table_name(table_name)
                  .schema(&schema_)
                  .add_range_partition_split(split_row.release())
                  .set_range_partition_columns({ key_column_name_ })
                  .num_replicas(num_replicas)
                  .Create());
    return Status::OK();
  }

  unique_ptr<KuduInsert> BuildTestRow(KuduTable* table, int index) {
    unique_ptr<KuduInsert> insert(table->NewInsert());
    KuduPartialRow* row = insert->mutable_row();
    CHECK_OK(row->SetInt32(0, index));
    CHECK_OK(row->SetInt32(1, index * 2));
    CHECK_OK(row->SetStringCopy(2, Slice(StringPrintf("hello %d", index))));
    return insert;
  }

  // Inserts given number of tests rows into the default test table
  // in the context of the specified session.
  Status InsertTestRows(KuduClient* client, KuduTable* table,
                        int num_rows, int first_row = 0) {
    shared_ptr<KuduSession> session = client->NewSession();
    RETURN_NOT_OK(session->SetFlushMode(KuduSession::AUTO_FLUSH_BACKGROUND));
    session->SetTimeoutMillis(60000);
    for (int i = first_row; i < num_rows + first_row; ++i) {
      unique_ptr<KuduInsert> insert(BuildTestRow(table, i));
      RETURN_NOT_OK(session->Apply(insert.release()));
    }
    RETURN_NOT_OK(session->Flush());
    return Status::OK();
  }

  Status GetRowCount(KuduTable* table, KuduScanner::ReadMode read_mode,
                     uint64_t ts, size_t* row_count) {
    return GetRowCount(table, read_mode, KuduClient::LEADER_ONLY, ts, row_count);
  }

  Status GetRowCount(KuduTable* table, KuduScanner::ReadMode read_mode,
                     KuduClient::ReplicaSelection selection, uint64_t ts,
                     size_t* row_count) {
    KuduScanner scanner(table);
    RETURN_NOT_OK(scanner.SetReadMode(read_mode));
    RETURN_NOT_OK(scanner.SetSelection(selection));
    if (read_mode == KuduScanner::READ_AT_SNAPSHOT && ts != 0) {
      RETURN_NOT_OK(scanner.SetSnapshotRaw(ts + 1));
    }
    RETURN_NOT_OK(CountRowsWithRetries(&scanner, row_count));
    return Status::OK();
  }

  Status GetTabletIdForKey(int32_t key_value, string* tablet_id) {
    if (!tablet_id) {
      return Status::InvalidArgument("null output string");
    }
    const int32_t key_value_begin = key_value;
    const int32_t key_value_end = key_value_begin + 1;

    unique_ptr<KuduPartialRow> split_row_start(schema_.NewRow());
    RETURN_NOT_OK(split_row_start->SetInt32(0, key_value_begin));
    string partition_key_start;
    RETURN_NOT_OK(split_row_start->EncodeRowKey(&partition_key_start));

    unique_ptr<KuduPartialRow> split_row_end(schema_.NewRow());
    RETURN_NOT_OK(split_row_end->SetInt32(0, key_value_end));
    string partition_key_end;
    RETURN_NOT_OK(split_row_end->EncodeRowKey(&partition_key_end));

    GetTableLocationsRequestPB req;
    req.mutable_table()->set_table_name(table_name_);
    req.set_partition_key_start(partition_key_start);
    req.set_partition_key_end(partition_key_end);
    master::CatalogManager* catalog =
        cluster_->mini_master()->master()->catalog_manager();
    GetTableLocationsResponsePB resp;
    CatalogManager::ScopedLeaderSharedLock l(catalog);
    RETURN_NOT_OK(l.first_failed_status());
    RETURN_NOT_OK(catalog->GetTableLocations(&req, &resp, /*user=*/none));
    if (resp.tablet_locations_size() < 1) {
      return Status::NotFound(Substitute("$0: no tablets for key", key_value));
    }
    if (resp.tablet_locations_size() > 1) {
      return Status::IllegalState(
          Substitute("$0: multiple tablet servers for key", key_value));
    }
    *tablet_id = resp.tablet_locations(0).tablet_id();

    return Status::OK();
  }

  Status FindPeerForTablet(const string& tablet_id,
                           scoped_refptr<TabletReplica>* replica) {
    bool found = false;
    for (size_t i = 0; i < num_tablet_servers_; ++i) {
      MiniTabletServer* mts = cluster_->mini_tablet_server(i);
      TabletServer* ts = mts->server();

      scoped_refptr<TabletReplica> r;
      if (!ts->tablet_manager()->LookupTablet(tablet_id, &r)) {
        // Not this one, continue.
        continue;
      }
      replica->swap(r);
      found = true;
      break;
    }
    if (!found) {
      return Status::NotFound(
          Substitute("$0: cannot find replica for tablet"), tablet_id);
    }
    return Status::OK();
  }

  Status UpdateClockForTabletHostingKey(int32_t key, const MonoDelta& offset) {
    string tablet_id;
    RETURN_NOT_OK(GetTabletIdForKey(key, &tablet_id));
    scoped_refptr<TabletReplica> r;
    RETURN_NOT_OK(FindPeerForTablet(tablet_id, &r));

    HybridClock* clock = CHECK_NOTNULL(dynamic_cast<HybridClock*>(r->clock()));
    UpdateClock(clock, offset);
    return Status::OK();
  }

  const size_t num_tablet_servers_;
  const string table_name_;
  const string key_column_name_;
  const int key_split_value_;
  KuduSchema schema_;
};

// This is a test that exposes the necessity of propagating timestamp
// between reads (scans) if consistent results are desired.
//
// Let T1, T2 be reads from the same client where T2 starts after the response
// from T1 is received and neither are assigned timestamps by the client.
// It might be the case where T2’s observed value actually precedes T1’s value
// in the row history if T1 and T2 are performed in different servers,
// as T2 can be assigned a timestamp that is lower than T1.
//
// The scenario to expose inconsistency when not propagating the timestamp
// for the scan operations:
//
//   * Set the flag to use the mock wall clock.
//
//   * Start two mini cluster tservers and create a table with two tablets
//     ({Ta, Tb} order matters), single replica.
//
//   * Advance the clock in tablet Ta's tserver by some amount.
//
//   * Write a row to the tablet Ta, discard the client.
//
//   * With a new client, read the row from the Ta tablet. Here we are using
//     READ_LATEST read mode because with READ_AT_SNAPSHOT mode the read
//     operation would block and wait for the clock to advance (which in this
//     case is possible only if doing that manually).
//     Then write a row to tablet Tb. Take note of write timestamp and
//     discard the client.
//
//   * Now scan both tablets using READ_AT_SNAPSHOT mode and the timestamp
//     of the write to the Tb tablet.
//     Since the write to the Tb tablet followed the write to the Ta tablet
//     (quite literally, we read those rows from the Ta tablet before writing
//     to the Tb tablet), we should see all the rows written.
//
//     However, that would not be the case if the scan timestamp from the Ta's
//     scan were not propagated to the Tb's write: Tb server's time is lagging
//     behind Ta server's time, and scanning at Tb's write time would not
//     include the rows inserted into Ta.
//
TEST_F(ConsistencyITest, TestTimestampPropagationFromScans) {
  uint64_t ts_a;
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    ASSERT_OK(CreateTable(client.get(), table_name_));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    // Advance tablet server's clock hosting the first key range
    // (i.e. for the row which is about to be inserted below).
    ASSERT_OK(UpdateClockForTabletHostingKey(
        0, MonoDelta::FromMilliseconds(100)));

    // Insert data into the first tablet (a.k.a. Ta).
    const int rows_num = key_split_value_;  // fill in the partition completely
    ASSERT_OK(InsertTestRows(client.get(), table.get(), rows_num, 0));
    size_t row_count;
    ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_LATEST, 0,
                          &row_count));
    ASSERT_EQ(key_split_value_, row_count);
    // Retrieve the latest observed timestamp.
    ts_a = client->GetLatestObservedTimestamp();
  }

  uint64_t ts_b;
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    // Running the prior version of the client code and using READ_AT_SNAPSHOT
    // for the scan would lead to unintentional propagating of the scan time
    // (i.e. Ta server's time) to the latest observed timestamp of the client.
    // Besides, as already mentioned in the comment for the test, the scan
    // in READ_AT_SNAPSHOT mode would block and wait for the clock to advance.
    size_t row_count;
    ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_LATEST, 0,
                          &row_count));
    // Check we see the first batch of inserted rows.
    ASSERT_EQ(key_split_value_, row_count);

    // Inserting data into the second tablet (a.k.a. Tb): using the second
    // key range partition.
    const int rows_num = key_split_value_;
    ASSERT_OK(InsertTestRows(client.get(), table.get(),
                             rows_num, key_split_value_));
    // Retrieve the latest observed timestamp.
    ts_b = client->GetLatestObservedTimestamp();
  }
  EXPECT_GT(ts_b, ts_a);

  // We are expecting to see all the inserted rows. It might not be the case
  // if the the timestamps were not propagated.
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    size_t row_count;
    ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_AT_SNAPSHOT, ts_b,
                          &row_count));
    // In total, inserted 2 * key_split_value_ rows and expecting to see
    // the total row count to reflect that: using snapshot timestamp that
    // is taken at insertng the second batch when the first batch already
    // was there.
    const size_t total_rows = 2UL * key_split_value_;
    ASSERT_EQ(total_rows, row_count);
  }
}

// Make sure the client propagates the timestamp for write operations.
//
// The idea of verification is simple:
//
//   * Let's get two tablet servers, where the clock of the first server
//     is ahead of the second one.
//
//   * Create a client object.
//
//   * Using the newly created client object, insert some data into the tablet
//     hosted by the first server.
//
//   * Record the client's latest observed timestamp.
//
//   * Using the same client object, insert some data into the tablet
//     hosted by the second server.
//
//   * Get the client's latest observed timestamp: it should be strictly greater
//     than the recorded timestamp.
//
//   * Make a full table scan in the READ_AT_SNAPSHOT mode at 'ts_ref'
//     timestamp: the scan should retrieve only the first row.
//
// If the client propates the timestamps, the second server should receive
// the recorded timestamp value in write request in the 'propagated_timestamp'
// field and adjust its clock first. After that it should perform the requested
// write operation. Since a write operation should always advance the server
// clock, the resulting timestamp returned to the client should be strictly
// greater than the propagated one.
TEST_F(ConsistencyITest, TestTimestampPropagationForWriteOps) {
  const int32_t offset_usec = FLAGS_max_clock_sync_error_usec;
  // Assuming the offset is specified as a positive number.
  ASSERT_GT(offset_usec, 0);
  // Need to have at least one row in the first partition starting with key 0.
  ASSERT_LE(1, key_split_value_);

  uint64_t ts_ref;
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    ASSERT_OK(CreateTable(client.get(), table_name_));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    // Advance tablet server's clock hosting the first key range
    // (i.e. for the row which is about to be inserted below).
    ASSERT_OK(UpdateClockForTabletHostingKey(
        0, MonoDelta::FromMicroseconds(offset_usec)));

    // Insert 1 row into the first tablet.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, 0));
    // Retrieve the latest observed timestamp.
    ts_ref = client->GetLatestObservedTimestamp();

    // Insert 1 row into the second tablet.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, key_split_value_));
    // Retrieve the latest observed timestamp.
    const uint64_t ts = client->GetLatestObservedTimestamp();

    // If the client propagates the timestamp with write operations,
    // the timestamp received from the second server should be greater
    // than the timestamp received from the first server.
    EXPECT_GT(ts, ts_ref);
  }

  // An additional check: scan the table at the 'ts_ref' timestamp and
  // make sure only the first row is visible.
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    size_t row_count;
    ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_AT_SNAPSHOT,
                          ts_ref, &row_count));
    EXPECT_EQ(1, row_count);
  }
}

// This is a test for KUDU-1189. It verifies that in case of a READ_AT_SNAPSHOT
// scan with unspecified snapshot timestamp, the scanner picks timestamp from
// the first server that the data is read from. If the scan spans multiple
// tablets, the timestamp picked when scanning the first tablet is then used
// when scanning following tablets.
//
// The idea of the test is simple: have a scan spanned across two tablets
// where the clocks of the corresponding tablet servers are skewed. The sequence
// of actions is as following:
//
//   1. Create a table which spans across two tablets.
//
//   2. Run the first scenario:
//      * Advance the clock of the second tablet's server.
//      * Create a client object and with it:
//        ** Insert a row into the first tablet.
//        ** Insert a row into the second tablet.
//      * Discard the client object.
//      * Create a new client object and perform a scan at READ_AT_SNAPSHOT
//        mode, no timestamp specified.
//      * Given the tight timings on the after-the-insert scan and difference in
//        server clocks, there should only one row in the result if the snapshot
//        timestamp is taken from the first server. Otherwise, if the snapshot
//        timestamp was taken from the second server, both rows would be visible
//        for the scan.
//      * Discard the client object.
//
//   3. Run the second scenario:
//      * Advance the clock of the first tablet's server, so the clock of the
//        first tablet is ahead of the clock of the second one.
//      * Create a client object and with it:
//        ** Insert an additional row into the first tablet.
//      * Discard the client object.
//      * Create a new client object and perform a scan at READ_AT_SNAPSHOT
//        mode, no timestamp specified.
//      * All the inserted rows should be visible to the scan because we
//        expect the snapshot timestamp to be taken from the first tablet
//        server. If the snapshot timestamp was taken from the second server,
//        given the tight timings on the scan following the prior insert into
//        the first tablet and difference in server clocks, not all rows would
//        be visible the the scan.
//
TEST_F(ConsistencyITest, TestSnapshotScanTimestampReuse) {
  const int32_t offset_usec = FLAGS_max_clock_sync_error_usec / 2;
  // Assuming the offset is specified as a positive number.
  ASSERT_GT(offset_usec, 0);
  // Need to have two rows in the first partition; the values start at 0.
  ASSERT_LT(2, key_split_value_);

  // Prepare the setup: create a proper disposition for tablet servers' clocks
  // and populate the table with appropriate data.
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    ASSERT_OK(CreateTable(client.get(), table_name_));
    // Advance second partition's tablet server clock.
    ASSERT_OK(UpdateClockForTabletHostingKey(
        key_split_value_, MonoDelta::FromMicroseconds(offset_usec)));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    // Insert a row into the first tablet.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, 0));
    // Insert a row into the second tablet.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, key_split_value_));
  }

  // Discarding the prior client object: if using it to perform scans, due
  // to the scan timestamp propagation the lagging tablet server's clock
  // would be advanced and it was not possible to distinguish between
  // the timestamps coming from the first and the second tablet servers.

  // Now, perform the scan at READ_AT_SNAPSHOT where a timestamp is not
  // specified: make sure the snapshot timestamp is taken from the first tablet
  // server among those the data was fetched from. For this scenario, perform
  // a scan which would try to fetch all the table's data
  // (i.e. make calls to all tablet servers which host table's data).
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    // Scan the table at a snapshot: let the servers pick the timestamp.
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    KuduScanner scanner(table.get());
    ASSERT_OK(scanner.SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
    const ScanConfiguration& cfg(scanner.data_->configuration());
    ASSERT_FALSE(cfg.has_snapshot_timestamp());

    size_t row_count;
    ASSERT_OK(CountRowsWithRetries(&scanner, &row_count));

    // At this point, we have inserted 2 rows in total, where the second row
    // was inserted into the tablet which server's clock was advanced
    // (i.e. shifted into the future). We are expecting to get the timestamp
    // for the scan from the first tablet server, so the second row should not
    // be visible at that timestamp: from the second tablet server's view,
    // it was inserted after the specified timestamp. Instead, if the timestamp
    // for the scan were sampled at the second server's clock, then both rows
    // would be visible to the scan.
    ASSERT_EQ(1UL, row_count);
    ASSERT_TRUE(cfg.has_snapshot_timestamp());
  }

  // Advance the clock of the first server even further, leaving the clock
  // of the second server behind. Also, insert an additional row into the first
  // tablet.
  {
    // Find the tablet for the first range to advance its server's clock.
    ASSERT_OK(UpdateClockForTabletHostingKey(
        0, MonoDelta::FromMicroseconds(2 * offset_usec)));
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    // Insert an additional row into the first tablet.
    // This is to check that the timestamp is taken from the first tablet
    // server: since now the clocks of both tablet servers are ahead of the
    // timestamps of the inserted rows so far, there would be no way to tell
    // which server's clock is used for the scan using the number of rows
    // returned by the scan. In either case, there will be two rows.
    //
    // Now, once we add a new row into the first tablet, given the big time
    // margin provided by the current clock offset, we should see different
    // outcomes from the subsequent scan:
    //   * if the timestamp is taken from the first server, there should be
    //     three rows in the result
    //   * if the timestamp is taken from the second server, there should be
    //     just two rows in the result
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, 1));
  }

  // Scan the table again and make sure the snapshot scan's timestamp is taken
  // from the first tablet server, as before. However, now the clock of the
  // first tablet server is ahead of the second tablet server's clock. If the
  // timestamp was taken from the second server, there would be 2 rows
  // in the result. The expected result is 3 rows, since the timestamp should
  // be taken from the first server.
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    // Scan the table at snapshot: let the servers pick the timestamp.
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    KuduScanner scanner(table.get());
    ASSERT_OK(scanner.SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
    const ScanConfiguration& cfg(scanner.data_->configuration());
    // Check the snapshot timestamp is unset -- it's a fresh object for a
    // READ_AT_SNAPSHOT scan where the snapshot timestamp is not specified
    // explicitly.
    ASSERT_FALSE(cfg.has_snapshot_timestamp());

    size_t row_count;
    ASSERT_OK(CountRowsWithRetries(&scanner, &row_count));

    // At this point, we have inserted 3 rows in total. Since the snapshot
    // timestamp is taken from the first server's clock, all 3 rows should be
    // visible to the scan at that timestamp. Given the tight timings on the
    // after-the-intsert scan and difference in server clocks, that would not be
    // the case if the snapshot was taken from the second server.
    ASSERT_EQ(3UL, row_count);
    // Check that the timestamp returned by the tablet server is set into the
    // scan configuration.
    ASSERT_TRUE(cfg.has_snapshot_timestamp());
  }
}

// Verify that the propagated timestamp from a serialized scan token
// makes its way into corresponding tablet servers while performing a scan
// operation built from the token.
//
// The real-world use-cases behind this test assume a Kudu client (C++/Java)
// can get scan tokens for a scan operation, serialize those and pass them
// to the other Kudu client (C++/Java). Since de-serializing a scan token
// propagates the latest observed timestamp, the latter client will have
// the latest observed timestamp set accordingly if it de-serializes those
// scan tokens into corresponding scan operations.
//
// The test scenario uses a table split into two tablets, each hosted by a
// tablet server. The clock of the first tablet server is shifted into the
// future. The first client inserts a row into the first tablet. Then it creates
// a scan token to retrieve some "related" data from the second
// tablet hosted by the second server. Now, another client receives the
// serialized scan token and runs corresponding READ_AT_SNAPSHOT scan
// with the specified timestamp to retrieve the data: it should observe
// a timestamp which is not less than the propagated timestamp
// encoded in the token.
TEST_F(ConsistencyITest, TestScanTokenTimestampPropagation) {
  const int32_t offset_usec = FLAGS_max_clock_sync_error_usec;

  // Need to have at least one row in the first partition with
  // values starting at 0.
  ASSERT_GE(key_split_value_, 1);

  {
    // Prepare the setup: create a proper disposition for tablet servers' clocks
    // and populate the table with initial data.
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    ASSERT_OK(CreateTable(client.get(), table_name_));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));

    // Insert a single row into the second tablet: it's necessary to get
    // non-empty scan in the verification phase of the test.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, key_split_value_));
  }

  uint64_t ts_ref;
  string scan_token;
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));

    // Advance the clock of the server hosting the first partition tablet.
    const int32_t row_key = 0;
    ASSERT_OK(UpdateClockForTabletHostingKey(
        row_key, MonoDelta::FromMicroseconds(offset_usec)));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    // Insert just a single row into the first tablet.
    ASSERT_OK(InsertTestRows(client.get(), table.get(), 1, row_key));
    ts_ref = client->GetLatestObservedTimestamp();

    // Create and serialize a scan token: the scan selects a row by its key
    // from the other tablet at the timestamp at which the first row was
    // inserted.
    vector<KuduScanToken*> tokens;
    ElementDeleter deleter(&tokens);
    KuduScanTokenBuilder builder(table.get());
    ASSERT_OK(builder.SetReadMode(KuduScanner::READ_AT_SNAPSHOT));
    unique_ptr<KuduPredicate> predicate(table->NewComparisonPredicate(
        key_column_name_,
        KuduPredicate::EQUAL,
        KuduValue::FromInt(key_split_value_)));
    ASSERT_OK(builder.AddConjunctPredicate(predicate.release()));
    ASSERT_OK(builder.Build(&tokens));
    ASSERT_EQ(1, tokens.size());
    ASSERT_OK(tokens[0]->Serialize(&scan_token));
  }

  // The other client: scan the second tablet using a scanner built from
  // the serialized scanner token. If the client propagates timestamp from the
  // de-serialized scan token, upon fetching a batch of rows the client
  // should observe timestamp not less than the reference propagated timestamp
  // encoded in the token.
  {
    shared_ptr<KuduClient> client;
    ASSERT_OK(cluster_->CreateClient(nullptr, &client));
    shared_ptr<KuduTable> table;
    ASSERT_OK(client->OpenTable(table_name_, &table));
    KuduScanner* scanner_raw;
    ASSERT_OK(KuduScanToken::DeserializeIntoScanner(client.get(), scan_token,
                                                    &scanner_raw));
    unique_ptr<KuduScanner> scanner(scanner_raw);
    ASSERT_OK(scanner->Open());
    ASSERT_TRUE(scanner->HasMoreRows());
    size_t row_count = 0;
    while (scanner->HasMoreRows()) {
      KuduScanBatch batch;
      ASSERT_OK(scanner->NextBatch(&batch));
      row_count += batch.NumRows();
      ASSERT_LE(ts_ref, client->GetLatestObservedTimestamp());
    }
    EXPECT_EQ(1, row_count);
  }
}

const KuduClient::ReplicaSelection replica_selectors[] = {
    KuduClient::LEADER_ONLY,
    KuduClient::CLOSEST_REPLICA,
    KuduClient::FIRST_REPLICA,
};

class ScanYourWritesParamTest :
    public ConsistencyITest,
    public ::testing::WithParamInterface<KuduClient::ReplicaSelection> {
};

// Verify that no matter which replica is selected, a single client could
// achieve read-your-writes on READ_YOUR_WRITES scan mode.
TEST_P(ScanYourWritesParamTest, Test) {
  const KuduClient::ReplicaSelection sel = GetParam();
  shared_ptr<KuduClient> client;
  ASSERT_OK(cluster_->CreateClient(nullptr, &client));
  ASSERT_OK(CreateTable(client.get(), table_name_, 3));
  shared_ptr<KuduTable> table;
  ASSERT_OK(client->OpenTable(table_name_, &table));

  // Insert multiple rows into the tablets.
  const uint64_t rows_to_insert = 20000;
  ASSERT_OK(InsertTestRows(client.get(), table.get(), rows_to_insert, 0));

  size_t row_count;
  ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_YOUR_WRITES,
                        sel, 0, &row_count));
  EXPECT_EQ(rows_to_insert, row_count);

  row_count = 0;
  ASSERT_OK(GetRowCount(table.get(), KuduScanner::READ_YOUR_WRITES,
                        sel, 0, &row_count));
  EXPECT_EQ(rows_to_insert, row_count);
}

INSTANTIATE_TEST_CASE_P(Params, ScanYourWritesParamTest,
                        testing::ValuesIn(replica_selectors));

class ScanYourWritesMultiClientsParamTest :
    public ConsistencyITest,
    public ::testing::WithParamInterface<KuduClient::ReplicaSelection> {
};

// This is a test that verifies, when multiple clients running
// simultaneously, a client can get read-your-writes and
// read-your-reads session guarantees using READ_YOUR_WRITES
// scan mode, no matter which replica is selected.
//
// Read-your-writes guarantees that a client can see all previous
// writes that it performed.
//
// Read-your-reads guarantees all subsequent reads to a given object
// "never return any previous values" regarding writes that other
// clients have done.
//
// The test scenario is as the following:
//
// 1) Have multiple clients running concurrently,
//
// 2) From the same client performs multiple writes and
//    multiple sets of scans. Each client
//    continuously performs inserts to a tablet, and then
//    performs a bunch of READ_YOUR_WRITES scans to all the
//    replicas that count the rows. The count of the rows
//    should never go down from the previous observed one.
TEST_P(ScanYourWritesMultiClientsParamTest, Test) {
  const int kNumThreads = 5;
  const int rows_to_insert = 1000;
  const int scans_to_perform = AllowSlowTests() ? 10 : 3;
  const KuduClient::ReplicaSelection sel = GetParam();

  shared_ptr<KuduClient> client;
  ASSERT_OK(cluster_->CreateClient(nullptr, &client));
  ASSERT_OK(CreateTable(client.get(), table_name_, 3));

  for (int run = 1; run <= 3; run++) {
    vector<thread> threads;
    for (int i = 0; i < kNumThreads; i++) {
      // TODO(adar): this is broken: each call to Next32() yields the same
      // value. Fixing it is non-trivial because 'first_row' is multiplied in
      // various code paths, causing it to overflow an int32. It just so happens
      // that the first Next32() using this particular seed generates a low
      // enough value to avoid overflow.
      Random rng(rows_to_insert * kNumThreads);
      uint32_t first_row = rng.Next32();
      threads.emplace_back([=]() {
        this->ScannerThread(sel, rows_to_insert, first_row, scans_to_perform);
      });
    }
    SleepFor(MonoDelta::FromMilliseconds(50));

    for (auto& t : threads) {
      t.join();
    }
  }
}

INSTANTIATE_TEST_CASE_P(Params, ScanYourWritesMultiClientsParamTest,
                        testing::ValuesIn(replica_selectors));
} // namespace client
} // namespace kudu