Cluster Read only option, Java 9+ ambitious method resolution issue with Properties and HashTable

app/controllers/Cluster.scala

@@ -93,6 +93,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
       , "jmxUser" -> optional(text)
       , "jmxPass" -> optional(text)
       , "jmxSsl" -> boolean
+      , "restrictOperations" -> boolean
       , "pollConsumers" -> boolean
       , "filterConsumers" -> boolean
       , "logkafkaEnabled" -> boolean
@@ -137,6 +138,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
       "jmxUser" -> optional(text),
       "jmxPass" -> optional(text),
       "jmxSsl" -> boolean,
+      "restrictOperations" -> boolean,
       "pollConsumers" -> boolean,
       "filterConsumers" -> boolean,
       "logkafkaEnabled" -> boolean,
@@ -185,6 +187,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
       ,false
       ,false
       ,false
+      ,false
       ,Option(defaultTuning)
       ,PLAINTEXT
       ,None
@@ -230,6 +233,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
             cc.jmxUser,
             cc.jmxPass,
             cc.jmxSsl,
+            cc.restrictOperations,
             cc.pollConsumers,
             cc.filterConsumers,
             cc.logkafkaEnabled,
@@ -258,6 +262,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
             clusterConfig.jmxUser,
             clusterConfig.jmxPass,
             clusterConfig.jmxSsl,
+            clusterConfig.restrictOperations,
             clusterConfig.pollConsumers,
             clusterConfig.filterConsumers,
             clusterConfig.tuning,
@@ -329,6 +334,7 @@ class Cluster (val cc: ControllerComponents, val kafkaManagerContext: KafkaManag
               clusterOperation.clusterConfig.jmxUser,
               clusterOperation.clusterConfig.jmxPass,
               clusterOperation.clusterConfig.jmxSsl,
+              clusterOperation.clusterConfig.restrictOperations,
               clusterOperation.clusterConfig.pollConsumers,
               clusterOperation.clusterConfig.filterConsumers,
               clusterOperation.clusterConfig.tuning,

app/kafka/manager/KafkaManager.scala

@@ -259,6 +259,7 @@ class KafkaManager(akkaConfig: Config) extends Logging {
                  jmxUser: Option[String],
                  jmxPass: Option[String],
                  jmxSsl: Boolean,
+                 restrictOperations: Boolean,
                  pollConsumers: Boolean,
                  filterConsumers: Boolean,
                  tuning: Option[ClusterTuning],
@@ -281,6 +282,7 @@ class KafkaManager(akkaConfig: Config) extends Logging {
       jmxUser = jmxUser,
       jmxPass = jmxPass,
       jmxSsl = jmxSsl,
+      restrictOperations = restrictOperations,
       pollConsumers = pollConsumers,
       filterConsumers = filterConsumers,
       logkafkaEnabled = logkafkaEnabled,
@@ -298,6 +300,7 @@ class KafkaManager(akkaConfig: Config) extends Logging {
                     jmxUser: Option[String],
                     jmxPass: Option[String],
                     jmxSsl: Boolean,
+                    restrictOperations: Boolean,
                     pollConsumers: Boolean,
                     filterConsumers: Boolean,
                     tuning: Option[ClusterTuning],
@@ -320,6 +323,7 @@ class KafkaManager(akkaConfig: Config) extends Logging {
       jmxUser = jmxUser,
       jmxPass = jmxPass,
       jmxSsl = jmxSsl,
+      restrictOperations = restrictOperations,
       pollConsumers = pollConsumers,
       filterConsumers = filterConsumers,
       logkafkaEnabled = logkafkaEnabled,

app/kafka/manager/actor/KafkaManagerActor.scala

@@ -489,6 +489,7 @@ class KafkaManagerActor(kafkaManagerConfig: KafkaManagerActorConfig)
         && newConfig.jmxPass == currentConfig.jmxPass
         && newConfig.jmxSsl == currentConfig.jmxSsl
         && newConfig.logkafkaEnabled == currentConfig.logkafkaEnabled
+        && newConfig.restrictOperations == currentConfig.restrictOperations
         && newConfig.pollConsumers == currentConfig.pollConsumers
         && newConfig.filterConsumers == currentConfig.filterConsumers
         && newConfig.activeOffsetCacheEnabled == currentConfig.activeOffsetCacheEnabled

app/kafka/manager/features/KMFeature.scala

@@ -20,6 +20,7 @@ sealed trait ClusterFeature extends KMFeature
 
 case object KMLogKafkaFeature extends ClusterFeature
 case object KMDeleteTopicFeature extends ClusterFeature
+case object KMRestrictedFeature extends ClusterFeature
 case object KMJMXMetricsFeature extends ClusterFeature
 case object KMDisplaySizeFeature extends ClusterFeature
 case object KMPollConsumersFeature extends ClusterFeature
@@ -73,6 +74,9 @@ object ClusterFeatures {
     if(clusterConfig.pollConsumers)
       buffer+=KMPollConsumersFeature
 
+    if(clusterConfig.restrictOperations)
+      buffer+=KMRestrictedFeature
+
     ClusterFeatures(buffer.toSet)
   }
 }

app/kafka/manager/model/model.scala

@@ -185,6 +185,7 @@ object ClusterConfig {
             , jmxUser: Option[String]
             , jmxPass: Option[String]
             , jmxSsl: Boolean
+            , restrictOperations:Boolean
             , pollConsumers: Boolean
             , filterConsumers: Boolean
             , logkafkaEnabled: Boolean = false
@@ -210,6 +211,7 @@ object ClusterConfig {
       , jmxUser
       , jmxPass
       , jmxSsl
+      , restrictOperations
       , pollConsumers
       , filterConsumers
       , logkafkaEnabled
@@ -223,10 +225,10 @@ object ClusterConfig {
   }
 
   def customUnapply(cc: ClusterConfig) : Option[(
-    String, String, String, Int, Boolean, Option[String], Option[String], Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Option[ClusterTuning], String, Option[String], Option[String])] = {
+    String, String, String, Int, Boolean, Option[String], Option[String], Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Option[ClusterTuning], String, Option[String], Option[String])] = {
     Some((
       cc.name, cc.version.toString, cc.curatorConfig.zkConnect, cc.curatorConfig.zkMaxRetry,
-      cc.jmxEnabled, cc.jmxUser, cc.jmxPass, cc.jmxSsl, cc.pollConsumers, cc.filterConsumers,
+      cc.jmxEnabled, cc.jmxUser, cc.jmxPass, cc.jmxSsl, cc.restrictOperations, cc.pollConsumers, cc.filterConsumers,
       cc.logkafkaEnabled, cc.activeOffsetCacheEnabled, cc.displaySizeEnabled, cc.tuning, cc.securityProtocol.stringId, cc.saslMechanism.map(_.stringId), cc.jaasConfig
       )
     )
@@ -264,6 +266,7 @@ object ClusterConfig {
       :: ("jmxUser" -> toJSON(config.jmxUser))
       :: ("jmxPass" -> toJSON(config.jmxPass))
       :: ("jmxSsl" -> toJSON(config.jmxSsl))
+      :: ("restrictOperations" -> toJSON(config.restrictOperations))
       :: ("pollConsumers" -> toJSON(config.pollConsumers))
       :: ("filterConsumers" -> toJSON(config.filterConsumers))
       :: ("logkafkaEnabled" -> toJSON(config.logkafkaEnabled))
@@ -290,6 +293,7 @@ object ClusterConfig {
           val jmxUser = fieldExtended[Option[String]]("jmxUser")(json)
           val jmxPass = fieldExtended[Option[String]]("jmxPass")(json)
           val jmxSsl = fieldExtended[Boolean]("jmxSsl")(json)
+          val restrictOperations = fieldExtended[Boolean]("restrictOperations")(json)
           val pollConsumers = fieldExtended[Boolean]("pollConsumers")(json)
           val filterConsumers = fieldExtended[Boolean]("filterConsumers")(json)
           val logkafkaEnabled = fieldExtended[Boolean]("logkafkaEnabled")(json)
@@ -310,6 +314,7 @@ object ClusterConfig {
             jmxUser.getOrElse(None),
             jmxPass.getOrElse(None),
             jmxSsl.getOrElse(false),
+            restrictOperations.getOrElse(false),
             pollConsumers.getOrElse(false),
             filterConsumers.getOrElse(true),
             logkafkaEnabled.getOrElse(false),
@@ -442,6 +447,7 @@ case class ClusterConfig (name: String
                           , jmxUser: Option[String]
                           , jmxPass: Option[String]
                           , jmxSsl: Boolean
+                          , restrictOperations:Boolean
                           , pollConsumers: Boolean
                           , filterConsumers: Boolean
                           , logkafkaEnabled: Boolean

app/kafka/manager/utils/logkafka81/LogConfig.scala

@@ -173,7 +173,8 @@ object LogConfig extends LogkafkaNewConfigs {
    */
   def fromProps(defaults: Properties, overrides: Properties): LogConfig = {
     val props = new Properties(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    overrides.forEach((k, v) => props.put(k, v))
     fromProps(props)
   }
 

app/kafka/manager/utils/logkafka82/LogConfig.scala

@@ -173,7 +173,8 @@ object LogConfig extends LogkafkaNewConfigs {
    */
   def fromProps(defaults: Properties, overrides: Properties): LogConfig = {
     val props = new Properties(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    overrides.forEach((k, v) => props.put(k, v))
     fromProps(props)
   }
 

app/kafka/manager/utils/zero10/LogConfig.scala

@@ -309,8 +309,9 @@ object LogConfig extends TopicConfigs {
     */
   def fromProps(defaults: java.util.Map[_ <: Object, _ <: Object], overrides: Properties): LogConfig = {
     val props = new Properties()
-    props.putAll(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    defaults.forEach((k, v) => props.put(k, v))
+    overrides.forEach((k, v) => props.put(k, v))
     LogConfig(props)
   }
 

app/kafka/manager/utils/zero11/LogConfig.scala

@@ -269,8 +269,9 @@ object LogConfig extends TopicConfigs {
     */
   def fromProps(defaults: java.util.Map[_ <: Object, _ <: Object], overrides: Properties): LogConfig = {
     val props = new Properties()
-    props.putAll(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    defaults.forEach((k, v) => props.put(k, v))
+    overrides.forEach((k, v) => props.put(k, v))
     LogConfig(props)
   }
 

app/kafka/manager/utils/zero81/LogConfig.scala

@@ -114,7 +114,8 @@ object LogConfig extends TopicConfigs {
    */
   def fromProps(defaults: Properties, overrides: Properties): LogConfig = {
     val props = new Properties(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    overrides.forEach((k, v) => props.put(k, v))
     fromProps(props)
   }
 

app/kafka/manager/utils/zero82/LogConfig.scala

@@ -181,7 +181,8 @@ object LogConfig extends TopicConfigs {
    */
   def fromProps(defaults: Properties, overrides: Properties): LogConfig = {
     val props = new Properties(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    overrides.forEach((k, v) => props.put(k, v))
     fromProps(props)
   }
 

app/kafka/manager/utils/zero90/LogConfig.scala

@@ -173,8 +173,9 @@ object LogConfig extends TopicConfigs {
     */
   def fromProps(defaults: java.util.Map[_ <: Object, _ <: Object], overrides: Properties): LogConfig = {
     val props = new Properties()
-    props.putAll(defaults)
-    props.putAll(overrides)
+    //to handle to Scala bug with Java 9+ versions https://github.com/scala/bug/issues/10418
+    defaults.forEach((k, v) => props.put(k, v))
+    overrides.forEach((k, v) => props.put(k, v))
     LogConfig(props)
   }
 

app/models/form/ClusterOperation.scala

@@ -39,6 +39,7 @@ object ClusterOperation {
             , jmxUser: Option[String]
             , jmxPass: Option[String]
             , jmxSsl: Boolean
+            , restrictOperations: Boolean
             , pollConsumers: Boolean
             , filterConsumers: Boolean
             , logkafkaEnabled: Boolean
@@ -50,14 +51,14 @@ object ClusterOperation {
             , jaasConfig: Option[String]
            ): ClusterOperation = {
     ClusterOperation(operation,ClusterConfig(name, version, zkHosts, zkMaxRetry, jmxEnabled, jmxUser, jmxPass, jmxSsl,
-      pollConsumers, filterConsumers, logkafkaEnabled, activeOffsetCacheEnabled, displaySizeEnabled, tuning, securityProtocol, saslMechanism, jaasConfig))
+      restrictOperations,pollConsumers, filterConsumers, logkafkaEnabled, activeOffsetCacheEnabled, displaySizeEnabled, tuning, securityProtocol, saslMechanism, jaasConfig))
   }
 
-  def customUnapply(co: ClusterOperation) : Option[(String, String, String, String, Int, Boolean, Option[String], Option[String], Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Option[ClusterTuning], String, Option[String], Option[String])] = {
+  def customUnapply(co: ClusterOperation) : Option[(String, String, String, String, Int, Boolean, Option[String], Option[String], Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Boolean, Option[ClusterTuning], String, Option[String], Option[String])] = {
     Option((co.op.toString, co.clusterConfig.name, co.clusterConfig.version.toString,
             co.clusterConfig.curatorConfig.zkConnect, co.clusterConfig.curatorConfig.zkMaxRetry,
             co.clusterConfig.jmxEnabled, co.clusterConfig.jmxUser, co.clusterConfig.jmxPass, co.clusterConfig.jmxSsl,
-            co.clusterConfig.pollConsumers, co.clusterConfig.filterConsumers, co.clusterConfig.logkafkaEnabled,
+      co.clusterConfig.restrictOperations,co.clusterConfig.pollConsumers, co.clusterConfig.filterConsumers, co.clusterConfig.logkafkaEnabled,
             co.clusterConfig.activeOffsetCacheEnabled, co.clusterConfig.displaySizeEnabled, co.clusterConfig.tuning, co.clusterConfig.securityProtocol.stringId,
             co.clusterConfig.saslMechanism.map(_.stringId),
             co.clusterConfig.jaasConfig))

app/models/navigation/Menus.scala

@@ -6,7 +6,7 @@
 package models.navigation
 
 import features.{KMTopicManagerFeature, KMClusterManagerFeature, KMPreferredReplicaElectionFeature, KMReassignPartitionsFeature, ApplicationFeatures}
-import kafka.manager.features.{KMLogKafkaFeature, ClusterFeatures}
+import kafka.manager.features.{KMLogKafkaFeature, ClusterFeatures, KMRestrictedFeature}
 
 /**
  * @author hiral
@@ -27,11 +27,11 @@ class Menus(implicit applicationFeatures: ApplicationFeatures) {
     Option(Menu("Cluster", items, None))
   }
 
-  private[this] def topicMenu(cluster: String) : Option[Menu] = {
+  private[this] def topicMenu(cluster: String, clusterFeatures: ClusterFeatures) : Option[Menu] = {
     val defaultItems = IndexedSeq("List".clusterRouteMenuItem(cluster))
 
     val items = {
-      if(applicationFeatures.features(KMTopicManagerFeature))
+      if(applicationFeatures.features(KMTopicManagerFeature) && !clusterFeatures.features(KMRestrictedFeature))
         defaultItems.+:("Create".clusterRouteMenuItem(cluster))
       else
         defaultItems
@@ -75,7 +75,7 @@ class Menus(implicit applicationFeatures: ApplicationFeatures) {
     IndexedSeq(
       clusterMenu(cluster),
       brokersMenu(cluster),
-      topicMenu(cluster),
+      topicMenu(cluster, clusterFeatures),
       preferredReplicaElectionMenu(cluster),
       reassignPartitionsMenu(cluster),
       consumersMenu(cluster),

app/views/cluster/addCluster.scala.html

@@ -31,6 +31,7 @@
         @b4.text(form("jmxUser"), '_label -> "JMX Auth Username")
         @b4.text(form("jmxPass"), '_label -> "JMX Auth Password")
         @b4.checkbox(form("jmxSsl"), '_text -> "JMX with SSL")
+        @b4.checkbox(form("restrictOperations"), '_text -> "Restrict the Operations(Non-Disruptive View only mode)")
         @b4.checkbox(form("logkafkaEnabled"), '_text -> "Enable Logkafka")
         @b4.checkbox(form("pollConsumers"), '_text -> "Poll consumer information (Not recommended for large # of consumers if ZK is used for offsets tracking on older Kafka versions)")
         @b4.checkbox(form("filterConsumers"), '_text -> "Filter out inactive consumers")

app/views/cluster/updateCluster.scala.html

@@ -34,6 +34,7 @@
         @b4.text(form("jmxUser"), '_label -> "JMX Auth Username")
         @b4.text(form("jmxPass"), '_label -> "JMX Auth Password")
         @b4.checkbox(form("jmxSsl"), '_text -> "JMX with SSL")
+        @b4.checkbox(form("restrictOperations"), '_text -> "Restrict the Operations(Non-Disruptive View only mode)")
         @b4.checkbox(form("pollConsumers"), '_text -> "Poll consumer information (Not recommended for large # of consumers if ZK is used for offsets tracking on older Kafka versions)")
         @b4.checkbox(form("filterConsumers"), '_text -> "Filter out inactive consumers")
         @b4.checkbox(form("logkafkaEnabled"), '_text -> "Enable Logkafka")

app/views/topic/topicList.scala.html

@@ -37,6 +37,32 @@
     }
 }
 
+@renderOperations = {
+
+@if(errorOrTopics.fold(err=>false,tl=>tl.list.headOption.map(opt => opt._2.map(ti => ti.clusterContext.config.restrictOperations && ti.clusterContext.config.name.equals(cluster)).getOrElse(false)).getOrElse(false))){
+    <div class="alert alert-info" role="alert">
+        Operations are restricted for @cluster Cluster (Enable at the cluster configurations)
+    </div>
+} else {
+    <table class="table">
+        <tr>
+            @features.app(features.KMReassignPartitionsFeature) {
+                <td>
+                    <a href="@routes.ReassignPartitions.confirmMultipleAssignments(cluster)" class="submit-button btn btn-primary" role="button">Generate Partition Assignments</a>
+                </td>
+                <td>
+                    <a href="@routes.ReassignPartitions.runMultipleAssignments(cluster)" class="submit-button btn btn-primary" role="button">Run Partition Assignments</a>
+                </td>
+            }
+            <td>
+                <a href="@routes.Topic.addPartitionsToMultipleTopics(cluster)" class="submit-button btn btn-primary" role="button">Add Partitions</a>
+            </td>
+        </tr>
+        </tbody>
+    </table>
+}
+}
+
 @main(
     "Topic List",
     menu = theMenu,
@@ -48,22 +74,7 @@
             <div class="card-header"><h4>Operations</h4></div>
             <div class="card-body">
             <div class="col-md-6">
-            <table class="table">
-                <tr>
-                    @features.app(features.KMReassignPartitionsFeature) {
-                    <td>
-                        <a href="@routes.ReassignPartitions.confirmMultipleAssignments(cluster)" class="submit-button btn btn-primary" role="button">Generate Partition Assignments</a>
-                    </td>
-                    <td>
-                        <a href="@routes.ReassignPartitions.runMultipleAssignments(cluster)" class="submit-button btn btn-primary" role="button">Run Partition Assignments</a>
-                    </td>
-                    }
-                    <td>
-                        <a href="@routes.Topic.addPartitionsToMultipleTopics(cluster)" class="submit-button btn btn-primary" role="button">Add Partitions</a>
-                    </td>
-                </tr>
-                </tbody>
-            </table>
+            @renderOperations
             </div>
             </div>
         </div>