KUDU-2514 Part 1: Support extra config for table.

We have thousands of tables in the Kudu cluster. It's hard to set a
uniform configuration for all tables. So, we support extra configuration
properties that are customizable on a per-table basis.

Note: In this patch, we implemented the framework, and the Java API. The
configuration item only validate framework & Java API. It doesn't actually
affect the tablet.

Change-Id: I0514507dca95602a97e954d1db464b907e073aae
Reviewed-on: http://gerrit.cloudera.org:8080/12468
Tested-by: Kudu Jenkins
Reviewed-by: Andrew Wong <[email protected]>

java/kudu-client/src/main/java/org/apache/kudu/client/AlterTableOptions.java

@@ -18,6 +18,7 @@
 package org.apache.kudu.client;
 
 import java.util.EnumSet;
+import java.util.Map;
 
 import static org.apache.kudu.ColumnSchema.CompressionAlgorithm;
 import static org.apache.kudu.ColumnSchema.Encoding;
@@ -380,6 +381,20 @@ public AlterTableOptions changeComment(String name, String comment) {
     return this;
   }
 
+  /**
+   * Change the table's extra configuration properties.
+   * These configuration properties will be merged into existing configuration properties.
+   *
+   * If the value of the kv pair is empty, the property will be unset.
+   *
+   * @param extraConfig the table's extra configuration properties
+   * @return this instance
+   */
+  public AlterTableOptions alterExtraConfigs(Map<String, String> extraConfig) {
+    pb.putAllNewExtraConfigs(extraConfig);
+    return this;
+  }
+
   /**
    * Whether to wait for the table to be fully altered before this alter
    * operation is considered to be finished.

java/kudu-client/src/main/java/org/apache/kudu/client/AsyncKuduClient.java

@@ -369,7 +369,7 @@ private AsyncKuduClient(AsyncKuduClientBuilder b) {
     this.channelFactory = b.createChannelFactory();
     this.masterAddresses = b.masterAddresses;
     this.masterTable = new KuduTable(this, MASTER_TABLE_NAME_PLACEHOLDER,
-        MASTER_TABLE_NAME_PLACEHOLDER, null, null, 1);
+        MASTER_TABLE_NAME_PLACEHOLDER, null, null, 1, null);
     this.defaultOperationTimeoutMs = b.defaultOperationTimeoutMs;
     this.defaultAdminOperationTimeoutMs = b.defaultAdminOperationTimeoutMs;
     this.statisticsDisabled = b.statisticsDisabled;
@@ -796,7 +796,8 @@ public KuduTable call(GetTableSchemaResponse resp) throws Exception {
             resp.getTableId(),
             resp.getSchema(),
             resp.getPartitionSchema(),
-            resp.getNumReplicas());
+            resp.getNumReplicas(),
+            resp.getExtraConfig());
       }
     });
   }

java/kudu-client/src/main/java/org/apache/kudu/client/CreateTableOptions.java

@@ -18,6 +18,7 @@
 package org.apache.kudu.client;
 
 import java.util.List;
+import java.util.Map;
 
 import com.google.common.base.Preconditions;
 import com.google.common.base.Strings;
@@ -190,6 +191,19 @@ public CreateTableOptions setNumReplicas(int numReplicas) {
     return this;
   }
 
+  /**
+   * Sets the table's extra configuration properties.
+   *
+   * If the value of the kv pair is empty, the property will be ignored.
+   *
+   * @param extraConfig the table's extra configuration properties
+   * @return this instance
+   */
+  public CreateTableOptions setExtraConfigs(Map<String, String> extraConfig) {
+    pb.putAllExtraConfigs(extraConfig);
+    return this;
+  }
+
   /**
    * Whether to wait for the table to be fully created before this create
    * operation is considered to be finished.

java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaRequest.java

@@ -99,7 +99,8 @@ Pair<GetTableSchemaResponse, Object> deserialize(CallResponse callResponse,
         respBuilder.getTableName(),
         respBuilder.getNumReplicas(),
         ProtobufHelper.pbToPartitionSchema(respBuilder.getPartitionSchema(), schema),
-        respBuilder.hasAuthzToken() ? respBuilder.getAuthzToken() : null);
+        respBuilder.hasAuthzToken() ? respBuilder.getAuthzToken() : null,
+        respBuilder.getExtraConfigsMap());
     return new Pair<GetTableSchemaResponse, Object>(
         response, respBuilder.hasError() ? respBuilder.getError() : null);
   }

java/kudu-client/src/main/java/org/apache/kudu/client/GetTableSchemaResponse.java

@@ -22,6 +22,8 @@
 import org.apache.kudu.security.Token.SignedTokenPB;
 import org.apache.kudu.Schema;
 
+import java.util.Map;
+
 @InterfaceAudience.Private
 public class GetTableSchemaResponse extends KuduRpcResponse {
 
@@ -31,6 +33,7 @@ public class GetTableSchemaResponse extends KuduRpcResponse {
   private final String tableName;
   private final int numReplicas;
   private final SignedTokenPB authzToken;
+  private final Map<String, String> extraConfig;
 
   /**
    * @param elapsedMillis Time in milliseconds since RPC creation to now
@@ -41,6 +44,7 @@ public class GetTableSchemaResponse extends KuduRpcResponse {
    * @param numReplicas the table's replication factor
    * @param partitionSchema the table's partition schema
    * @param authzToken an authorization token for use with this table
+   * @param extraConfig the table's extra configuration properties
    */
   GetTableSchemaResponse(long elapsedMillis,
                          String tsUUID,
@@ -49,14 +53,16 @@ public class GetTableSchemaResponse extends KuduRpcResponse {
                          String tableName,
                          int numReplicas,
                          PartitionSchema partitionSchema,
-                         SignedTokenPB authzToken) {
+                         SignedTokenPB authzToken,
+                         Map<String, String> extraConfig) {
     super(elapsedMillis, tsUUID);
     this.schema = schema;
     this.partitionSchema = partitionSchema;
     this.tableId = tableId;
     this.tableName = tableName;
     this.numReplicas = numReplicas;
     this.authzToken = authzToken;
+    this.extraConfig = extraConfig;
   }
 
   /**
@@ -106,4 +112,12 @@ public int getNumReplicas() {
   public SignedTokenPB getAuthzToken() {
     return authzToken;
   }
+
+  /**
+   * Get the table's extra configuration properties.
+   * @return the table's extra configuration properties
+   */
+  public Map<String, String> getExtraConfig() {
+    return extraConfig;
+  }
 }

java/kudu-client/src/main/java/org/apache/kudu/client/KuduTable.java

@@ -19,6 +19,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 
 import com.google.common.base.Predicates;
 import com.google.common.collect.Iterators;
@@ -48,6 +49,7 @@ public class KuduTable {
   private final String name;
   private final String tableId;
   private final int numReplicas;
+  private final Map<String, String> extraConfig;
 
   /**
    * Package-private constructor, use {@link KuduClient#openTable(String)} to get an instance.
@@ -57,15 +59,18 @@ public class KuduTable {
    * @param schema this table's schema
    * @param partitionSchema this table's partition schema
    * @param numReplicas this table's replication factor
+   * @param extraConfig this table's extra configuration properties
    */
   KuduTable(AsyncKuduClient client, String name, String tableId,
-            Schema schema, PartitionSchema partitionSchema, int numReplicas) {
+            Schema schema, PartitionSchema partitionSchema, int numReplicas,
+            Map<String, String> extraConfig) {
     this.schema = schema;
     this.partitionSchema = partitionSchema;
     this.client = client;
     this.name = name;
     this.tableId = tableId;
     this.numReplicas = numReplicas;
+    this.extraConfig = extraConfig;
   }
 
   /**
@@ -113,6 +118,14 @@ public int getNumReplicas() {
     return numReplicas;
   }
 
+  /**
+   * Get this table's extra configuration properties.
+   * @return this table's extra configuration properties
+   */
+  public Map<String, String> getExtraConfig() {
+    return extraConfig;
+  }
+
   /**
    * Get the async client that created this instance.
    * @return an async kudu client

java/kudu-client/src/test/java/org/apache/kudu/client/TestAlterTable.java

@@ -28,6 +28,8 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.HashMap;
+import java.util.Map;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Lists;
@@ -479,4 +481,44 @@ public void testAlterRangeParitioningInvalid() throws KuduException {
     }
     assertEquals(100, countRowsInTable(table));
   }
+
+  @Test
+  public void testAlterExtraConfigs() throws Exception {
+    KuduTable table = createTable(ImmutableList.<Pair<Integer,Integer>>of());
+    insertRows(table, 0, 100);
+    assertEquals(100, countRowsInTable(table));
+
+    // 1. Check for expected defaults.
+    table = client.openTable(tableName);
+    Map<String, String> extraConfigs = table.getExtraConfig();
+    assertFalse(extraConfigs.containsKey("kudu.table.history_max_age_sec"));
+
+    // 2. Alter history max age second to 3600
+    Map<String, String> alterExtraConfigs = new HashMap<>();
+    alterExtraConfigs.put("kudu.table.history_max_age_sec", "3600");
+    client.alterTable(tableName, new AlterTableOptions().alterExtraConfigs(alterExtraConfigs));
+
+    table = client.openTable(tableName);
+    extraConfigs = table.getExtraConfig();
+    assertTrue(extraConfigs.containsKey("kudu.table.history_max_age_sec"));
+    assertEquals("3600", extraConfigs.get("kudu.table.history_max_age_sec"));
+
+    // 3. Alter history max age second to 7200
+    alterExtraConfigs = new HashMap<>();
+    alterExtraConfigs.put("kudu.table.history_max_age_sec", "7200");
+    client.alterTable(tableName, new AlterTableOptions().alterExtraConfigs(alterExtraConfigs));
+
+    table = client.openTable(tableName);
+    extraConfigs = table.getExtraConfig();
+    assertTrue(extraConfigs.containsKey("kudu.table.history_max_age_sec"));
+    assertEquals("7200", extraConfigs.get("kudu.table.history_max_age_sec"));
+
+    // 4. Reset history max age second to default
+    alterExtraConfigs = new HashMap<>();
+    alterExtraConfigs.put("kudu.table.history_max_age_sec", "");
+    client.alterTable(tableName, new AlterTableOptions().alterExtraConfigs(alterExtraConfigs));
+
+    table = client.openTable(tableName);
+    assertTrue(table.getExtraConfig().isEmpty());
+  }
 }

java/kudu-client/src/test/java/org/apache/kudu/client/TestAsyncKuduClient.java

@@ -172,7 +172,7 @@ public void testBadHostnames() throws Exception {
     // Test that a tablet full of unreachable replicas won't make us retry.
     try {
       KuduTable badTable = new KuduTable(asyncClient, "Invalid table name",
-          "Invalid table ID", null, null, 3);
+          "Invalid table ID", null, null, 3, null);
       asyncClient.discoverTablets(badTable, null, requestBatchSize,
                                   tabletLocations, new ArrayList<>(), 1000);
       fail("This should have failed quickly");

java/kudu-client/src/test/java/org/apache/kudu/client/TestKeyEncoding.java

@@ -113,7 +113,7 @@ public void testPrimaryKeys() {
     Schema schemaOneString =
         buildSchema(new ColumnSchema.ColumnSchemaBuilder("key", Type.STRING).key(true));
     KuduTable table = new KuduTable(null, "one", "one", schemaOneString,
-                                    defaultPartitionSchema(schemaOneString), 3);
+                                    defaultPartitionSchema(schemaOneString), 3, null);
     Insert oneKeyInsert = new Insert(table);
     PartialRow row = oneKeyInsert.getRow();
     row.addString("key", "foo");
@@ -123,7 +123,7 @@ public void testPrimaryKeys() {
         new ColumnSchema.ColumnSchemaBuilder("key", Type.STRING).key(true),
         new ColumnSchema.ColumnSchemaBuilder("key2", Type.STRING).key(true));
     KuduTable table2 = new KuduTable(null, "two", "two", schemaTwoString,
-                                     defaultPartitionSchema(schemaTwoString), 3);
+                                     defaultPartitionSchema(schemaTwoString), 3, null);
     Insert twoKeyInsert = new Insert(table2);
     row = twoKeyInsert.getRow();
     row.addString("key", "foo");
@@ -142,7 +142,7 @@ public void testPrimaryKeys() {
         new ColumnSchema.ColumnSchemaBuilder("key2", Type.STRING).key(true));
     PartitionSchema partitionSchemaIntString = defaultPartitionSchema(schemaIntString);
     KuduTable table3 = new KuduTable(null, "three", "three",
-        schemaIntString, partitionSchemaIntString, 3);
+        schemaIntString, partitionSchemaIntString, 3, null);
     Insert small = new Insert(table3);
     row = small.getRow();
     row.addInt("key", 20);

java/kudu-client/src/test/java/org/apache/kudu/client/TestKuduClient.java

@@ -47,6 +47,8 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
+import java.util.HashMap;
 import java.util.Random;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutorService;
@@ -184,6 +186,26 @@ public void testCreateTableTooManyColumns() throws Exception {
     }
   }
 
+  /**
+   * Test creating and deleting a table with extra-configs through a KuduClient.
+   */
+  @Test(timeout = 100000)
+  public void testCreateDeleteTableWitExtraConfigs() throws Exception {
+    // Check that we can create a table.
+    Map<String, String> extraConfigs = new HashMap<>();
+    extraConfigs.put("kudu.table.history_max_age_sec", "7200");
+
+    client.createTable(
+        TABLE_NAME,
+        basicSchema,
+        getBasicCreateTableOptions().setExtraConfigs(extraConfigs));
+
+    KuduTable table = client.openTable(TABLE_NAME);
+    extraConfigs = table.getExtraConfig();
+    assertTrue(extraConfigs.containsKey("kudu.table.history_max_age_sec"));
+    assertEquals("7200", extraConfigs.get("kudu.table.history_max_age_sec"));
+  }
+
   /*
    * Test the scanner behavior when a scanner is used beyond
    * the scanner ttl without calling keepAlive.

src/kudu/common/common.proto

@@ -438,3 +438,11 @@ message KeyRangePB {
   // Number of bytes in chunk.
   required uint64 size_bytes_estimates = 3;
 }
+
+message TableExtraConfigPB {
+  // Number of seconds to retain history for tablets in this table,
+  // including history required to perform diff scans and incremental
+  // backups. Reads initiated at a snapshot that is older than this
+  // age will be rejected. Equivalent to --tablet_history_max_age_sec.
+  optional int32 history_max_age_sec = 1;
+}