Remove obsolete deprecated client Config keys (#8994)

docs/distcache-blobstore.md

@@ -321,8 +321,8 @@ The default is 60 seconds, a value of -1 indicates to wait for ever.
 
 Additionally, if you want to access to secure hdfs blobstore, you also need to set the following configs.  
 ```
-storm.hdfs.login.keytab or blobstore.hdfs.keytab (deprecated)
-storm.hdfs.login.principal or blobstore.hdfs.principal (deprecated)
+storm.hdfs.login.keytab
+storm.hdfs.login.principal
 ```
 
 For example,

external/storm-blobstore-migration/README.md

@@ -59,8 +59,8 @@ To migrate blobs from nimbus, the following steps are necessary:
 2. Backup storm config
 3. Change the following settings in Nimbus' storm config:
    * blobstore.dir
-   * blobstore.hdfs.principal
-   * blobstore.hdfs.keytab
+   * storm.hdfs.login.principal
+   * storm.hdfs.login.keytab
    * blobstore.replication.factor
    * nimbus.blobstore.class
 4. Configure server so that the environment variable `STORM_EXT_CLASSPATH` includes whatever `HADOOP_CLASSPATH` contains when `storm nimbus` is run.
@@ -76,8 +76,8 @@ Supervisors can be upgraded by performing the following steps:
 1. Shut down the supervisor.
 2. Putting the following blobstore settings in place:
    * blobstore.dir
-   * blobstore.hdfs.principal
-   * blobstore.hdfs.keytab
+   * storm.hdfs.login.principal
+   * storm.hdfs.login.keytab
    * blobstore.replication.factor
    * supervisor.blobstore.class
 3. Kill all remaining worker processes (this is ugly)

storm-client/src/jvm/org/apache/storm/Config.java

@@ -1673,13 +1673,6 @@ public class Config extends HashMap<String, Object> {
     @IsListEntryCustom(entryValidatorClasses = {MetricReportersValidator.class})
     public static final String STORM_TOPOLOGY_METRICS_SYSTEM_REPORTERS = "storm.topology.metrics.system.reporters";
 
-    /**
-     * Configure the topology metrics reporters to be used on workers.
-     * Use {@link Config#TOPOLOGY_METRICS_REPORTERS} instead.
-     */
-    @Deprecated(forRemoval = true, since = "2.0.0")
-    @IsListEntryCustom(entryValidatorClasses = {MetricReportersValidator.class})
-    public static final String STORM_METRICS_REPORTERS = "storm.metrics.reporters";
 
     /**
      * What blobstore implementation the storm client should use.
@@ -1700,25 +1693,7 @@ public class Config extends HashMap<String, Object> {
      */
     @IsBoolean
     public static final String BLOBSTORE_CLEANUP_ENABLE = "blobstore.cleanup.enable";
-    /**
-     * principal for nimbus/supervisor to use to access secure hdfs for the blobstore.
-     * The format is generally "primary/instance@REALM", where "instance" field is optional.
-     * If the instance field of the principal is the string "_HOST", it will
-     * be replaced with the host name of the server the daemon is running on
-     * (by calling {@link #getBlobstoreHDFSPrincipal(Map conf)} method).
-     * Note: Use {@link Config#STORM_HDFS_LOGIN_PRINCIPAL} instead.
-     */
-    @Deprecated(forRemoval = true, since = "2.0.0")
-    @IsString
-    public static final String BLOBSTORE_HDFS_PRINCIPAL = "blobstore.hdfs.principal";
-    /**
-     * keytab for nimbus/supervisor to use to access secure hdfs for the blobstore.
-     *
-     * @Deprecated Use {@link Config#STORM_HDFS_LOGIN_KEYTAB} instead.
-     */
-    @Deprecated
-    @IsString
-    public static final String BLOBSTORE_HDFS_KEYTAB = "blobstore.hdfs.keytab";
+
     /**
      * Set replication factor for a blob in HDFS Blobstore Implementation.
      */
@@ -2640,39 +2615,19 @@ private static String substituteHostnameInPrincipal(String principal) throws Unk
         return principal;
     }
 
-    @Deprecated
-    @SuppressWarnings("checkstyle:AbbreviationAsWordInName")
-    public static String getBlobstoreHDFSPrincipal(Map conf) throws UnknownHostException {
-        return getHdfsPrincipal(conf);
-    }
-
     /**
      * Get the hostname substituted hdfs principal.
      *
      * @param conf the storm Configuration
      * @return the principal
-     * @throws UnknownHostException on UnknowHostException
+     * @throws UnknownHostException on UnknownHostException
      */
     public static String getHdfsPrincipal(Map<String, Object> conf) throws UnknownHostException {
-        String ret;
-
-        String blobstorePrincipal = (String) conf.get(Config.BLOBSTORE_HDFS_PRINCIPAL);
         String hdfsPrincipal = (String) conf.get(Config.STORM_HDFS_LOGIN_PRINCIPAL);
-        if (blobstorePrincipal == null && hdfsPrincipal == null) {
+        if (hdfsPrincipal == null) {
             return null;
-        } else if (blobstorePrincipal == null) {
-            ret = hdfsPrincipal;
-        } else if (hdfsPrincipal == null) {
-            LOG.warn("{} is used as the hdfs principal. Please use {} instead",
-                Config.BLOBSTORE_HDFS_PRINCIPAL, Config.STORM_HDFS_LOGIN_PRINCIPAL);
-            ret = blobstorePrincipal;
-        } else {
-            //both not null;
-            LOG.warn("Both {} and {} are set. Use {} only.",
-                Config.BLOBSTORE_HDFS_PRINCIPAL, Config.STORM_HDFS_LOGIN_PRINCIPAL, Config.STORM_HDFS_LOGIN_PRINCIPAL);
-            ret = hdfsPrincipal;
         }
-        return substituteHostnameInPrincipal(ret);
+        return substituteHostnameInPrincipal(hdfsPrincipal);
     }
 
     /**
@@ -2682,25 +2637,7 @@ public static String getHdfsPrincipal(Map<String, Object> conf) throws UnknownHo
      * @return the keytab
      */
     public static String getHdfsKeytab(Map<String, Object> conf) {
-        String ret;
-
-        String blobstoreKeyTab = (String) conf.get(Config.BLOBSTORE_HDFS_KEYTAB);
-        String hdfsKeyTab = (String) conf.get(Config.STORM_HDFS_LOGIN_KEYTAB);
-        if (blobstoreKeyTab == null && hdfsKeyTab == null) {
-            return null;
-        } else if (blobstoreKeyTab == null) {
-            ret = hdfsKeyTab;
-        } else if (hdfsKeyTab == null) {
-            LOG.warn("{} is used as the hdfs keytab. Please use {} instead",
-                Config.BLOBSTORE_HDFS_KEYTAB, Config.STORM_HDFS_LOGIN_KEYTAB);
-            ret = blobstoreKeyTab;
-        } else {
-            //both not null;
-            LOG.warn("Both {} and {} are set. Use {} only.",
-                Config.BLOBSTORE_HDFS_KEYTAB, Config.STORM_HDFS_LOGIN_KEYTAB, Config.STORM_HDFS_LOGIN_KEYTAB);
-            ret = hdfsKeyTab;
-        }
-        return ret;
+        return (String) conf.get(Config.STORM_HDFS_LOGIN_KEYTAB);
     }
 
 }

storm-client/test/jvm/org/apache/storm/utils/ConfigUtilsTest.java

@@ -99,36 +99,6 @@ public void getValueAsList_nonStringList() {
         assertEquals(expectedValue, ConfigUtils.getValueAsList(key, map));
     }
 
-    @Deprecated
-    @Test
-    public void getBlobstoreHDFSPrincipal() throws UnknownHostException {
-        Map<String, Object> conf = mockMap(Config.BLOBSTORE_HDFS_PRINCIPAL, "primary/_HOST@EXAMPLE.COM");
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), "primary/" +  Utils.localHostname() + "@EXAMPLE.COM");
-
-        String principal = "primary/_HOST_HOST@EXAMPLE.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-
-        principal = "primary/_HOST2@EXAMPLE.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-
-        principal = "_HOST/instance@EXAMPLE.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-
-        principal = "primary/instance@_HOST.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-
-        principal = "_HOST@EXAMPLE.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-
-        principal = "primary/instance@EXAMPLE.COM";
-        conf.put(Config.BLOBSTORE_HDFS_PRINCIPAL, principal);
-        assertEquals(Config.getBlobstoreHDFSPrincipal(conf), principal);
-    }
 
     @Test
     public void getHfdsPrincipal() throws UnknownHostException {

storm-server/src/main/java/org/apache/storm/daemon/nimbus/Nimbus.java

@@ -818,8 +818,8 @@
      */
     private static Map<String, Object> maskCredentialsForApi(Map<String, Object> conf) {
         Map<String, Object> masked = new HashMap<>(ConfigUtils.maskCredentials(conf));
         if (masked.get(BlowfishTupleSerializer.SECRET_KEY) instanceof String) {
             masked.put(BlowfishTupleSerializer.SECRET_KEY, "*****");
         }
         return masked;
     }
@@ -1138,7 +1138,7 @@
         cleanable.addAll(Utils.OR(state.heartbeatStorms(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.errorTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(store.storedTopoIds(), EMPTY_STRING_SET));
         cleanable.addAll(Utils.OR(state.backpressureTopologies(), EMPTY_STRING_LIST));
         cleanable.addAll(Utils.OR(state.idsOfTopologiesWithPrivateWorkerKeys(), EMPTY_STRING_SET));
         Set<String> delayedCleanable = getExpiredTopologyIds(cleanable, conf);
         delayedCleanable.removeAll(Utils.OR(state.activeStorms(), EMPTY_STRING_LIST));
@@ -1258,9 +1258,6 @@
         ret.put(Config.TOPOLOGY_WORKER_NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
         ret.put(Config.NIMBUS_THRIFT_CLIENT_USE_TLS, workerNimbusClientTlsEnabled);
 
-        if (!mergedConf.containsKey(Config.TOPOLOGY_METRICS_REPORTERS) && mergedConf.containsKey(Config.STORM_METRICS_REPORTERS)) {
-            ret.put(Config.TOPOLOGY_METRICS_REPORTERS, mergedConf.get(Config.STORM_METRICS_REPORTERS));
-        }
 
         // add any system metrics reporters to the topology metrics reporters
         if (conf.containsKey(Config.STORM_TOPOLOGY_METRICS_SYSTEM_REPORTERS)) {
@@ -3177,7 +3174,7 @@
                 state.teardownHeartbeats(topoId);
                 state.teardownTopologyErrors(topoId);
                 state.removeAllPrivateWorkerKeys(topoId);
                 state.removeBackpressure(topoId);
                 rmDependencyBlobsInTopology(topoId, stillReferenced);
                 forceDeleteTopoDistDir(topoId);
                 rmTopologyKeys(topoId);
@@ -3697,8 +3694,8 @@
                 waitForDesiredCodeReplication(totalConf, topoId);
                 state.setupHeatbeats(topoId, topoConf);
                 state.setupErrors(topoId, topoConf);
                 if (ObjectReader.getBoolean(totalConf.get(Config.TOPOLOGY_BACKPRESSURE_ENABLE), false)) {
                     state.setupBackpressure(topoId, topoConf);
                 }
                 notifyTopologyActionListener(topoName, "submitTopology");
                 TopologyStatus status = null;