diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
index ee034ec952ca..d82e058daf61 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ControllerMeter.java
@@ -68,7 +68,9 @@ public enum ControllerMeter implements AbstractMetrics.Meter {
   NUMBER_ADHOC_TASKS_SUBMITTED("adhocTasks", false),
   IDEAL_STATE_UPDATE_FAILURE("IdealStateUpdateFailure", false),
   IDEAL_STATE_UPDATE_RETRY("IdealStateUpdateRetry", false),
-  IDEAL_STATE_UPDATE_SUCCESS("IdealStateUpdateSuccess", false);
+  IDEAL_STATE_UPDATE_SUCCESS("IdealStateUpdateSuccess", false),
+  GROOVY_SECURITY_VIOLATIONS("exceptions", true);
+
 
 
   private final String _brokerMeterName;
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metrics/MinionMeter.java b/pinot-common/src/main/java/org/apache/pinot/common/metrics/MinionMeter.java
index 220fd7cb7d2b..96f2c87a6074 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metrics/MinionMeter.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metrics/MinionMeter.java
@@ -38,7 +38,8 @@ public enum MinionMeter implements AbstractMetrics.Meter {
   SEGMENT_BYTES_UPLOADED("bytes", false),
   RECORDS_PROCESSED_COUNT("rows", false),
   RECORDS_PURGED_COUNT("rows", false),
-  COMPACTED_RECORDS_COUNT("rows", false);
+  COMPACTED_RECORDS_COUNT("rows", false),
+  GROOVY_SECURITY_VIOLATIONS("exceptions", true);
 
   private final String _meterName;
   private final String _unit;
diff --git a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
index 6a53ddc57130..df5aece8716b 100644
--- a/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
+++ b/pinot-common/src/main/java/org/apache/pinot/common/metrics/ServerMeter.java
@@ -27,6 +27,7 @@
 public enum ServerMeter implements AbstractMetrics.Meter {
   QUERIES("queries", true),
   UNCAUGHT_EXCEPTIONS("exceptions", true),
+  GROOVY_SECURITY_VIOLATIONS("exceptions", true),
   REQUEST_DESERIALIZATION_EXCEPTIONS("exceptions", true),
   RESPONSE_SERIALIZATION_EXCEPTIONS("exceptions", true),
   SCHEDULING_TIMEOUT_EXCEPTIONS("exceptions", true),
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
index a0ce503d41a1..27f19eb93c38 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/BaseControllerStarter.java
@@ -122,6 +122,7 @@
 import org.apache.pinot.core.segment.processing.lifecycle.PinotSegmentLifecycleEventListenerManager;
 import org.apache.pinot.core.transport.ListenerConfig;
 import org.apache.pinot.core.util.ListenerConfigUtil;
+import org.apache.pinot.segment.local.function.GroovyFunctionEvaluator;
 import org.apache.pinot.segment.local.utils.TableConfigUtils;
 import org.apache.pinot.spi.config.table.TableConfig;
 import org.apache.pinot.spi.crypt.PinotCrypterFactory;
@@ -382,7 +383,8 @@ public ControllerConf getConfig() {
   }
 
   @Override
-  public void start() {
+  public void start()
+      throws Exception {
     LOGGER.info("Starting Pinot controller in mode: {}. (Version: {})", _controllerMode.name(), PinotVersion.VERSION);
     LOGGER.info("Controller configs: {}", new PinotAppConfigs(getConfig()).toJSONString());
     long startTimeMs = System.currentTimeMillis();
@@ -407,6 +409,12 @@ public void start() {
         break;
     }
 
+    // Initializing Groovy execution security
+    GroovyFunctionEvaluator.configureGroovySecurity(
+        _config.getProperty(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG));
+    GroovyFunctionEvaluator.setMetrics(_controllerMetrics, ControllerMeter.GROOVY_SECURITY_VIOLATIONS);
+
+
     ServiceStatus.setServiceStatusCallback(_helixParticipantInstanceId,
         new ServiceStatus.MultipleCallbackServiceStatusCallback(_serviceStatusCallbackList));
     _controllerMetrics.addTimedValue(ControllerTimer.STARTUP_SUCCESS_DURATION_MS,
diff --git a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotClusterConfigs.java b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotClusterConfigs.java
index f54751db16bf..07c4394f76ef 100644
--- a/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotClusterConfigs.java
+++ b/pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotClusterConfigs.java
@@ -55,6 +55,9 @@
 import org.apache.pinot.core.auth.Actions;
 import org.apache.pinot.core.auth.Authorize;
 import org.apache.pinot.core.auth.TargetType;
+import org.apache.pinot.segment.local.function.GroovyFunctionEvaluator;
+import org.apache.pinot.segment.local.function.GroovyStaticAnalyzerConfig;
+import org.apache.pinot.spi.utils.CommonConstants;
 import org.apache.pinot.spi.utils.JsonUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -168,4 +171,75 @@ public SuccessResponse deleteClusterConfig(
       throw new ControllerApplicationException(LOGGER, errStr, Response.Status.INTERNAL_SERVER_ERROR, e);
     }
   }
+
+  @GET
+  @Path("/cluster/configs/groovy/staticAnalyzerConfig")
+  @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_GROOVY_STATIC_ANALYZER_CONFIG)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Get the configuration for Groovy Static analysis",
+      notes = "Get the configuration for Groovy static analysis")
+  @ApiResponses(value = {
+      @ApiResponse(code = 200, message = "Success"),
+      @ApiResponse(code = 500, message = "Internal server error")
+  })
+  public GroovyStaticAnalyzerConfig getGroovyStaticAnalysisConfig() throws Exception {
+    HelixAdmin helixAdmin = _pinotHelixResourceManager.getHelixAdmin();
+    HelixConfigScope configScope = new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER)
+        .forCluster(_pinotHelixResourceManager.getHelixClusterName()).build();
+    Map<String, String> configs = helixAdmin.getConfig(configScope,
+        List.of(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG));
+    String json = configs.get(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG);
+    if (json != null) {
+      return GroovyStaticAnalyzerConfig.fromJson(json);
+    } else {
+      return null;
+    }
+  }
+
+
+  @POST
+  @Path("/cluster/configs/groovy/staticAnalyzerConfig")
+  @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.UPDATE_GROOVY_STATIC_ANALYZER_CONFIG)
+  @Authenticate(AccessType.UPDATE)
+  @ApiOperation(value = "Update Groovy static analysis configuration")
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiResponses(value = {
+      @ApiResponse(code = 200, message = "Success"),
+      @ApiResponse(code = 500, message = "Server error updating configuration")
+  })
+  public SuccessResponse setGroovyStaticAnalysisConfig(String body) throws Exception {
+    try {
+      HelixAdmin admin = _pinotHelixResourceManager.getHelixAdmin();
+      HelixConfigScope configScope =
+          new HelixConfigScopeBuilder(HelixConfigScope.ConfigScopeProperty.CLUSTER).forCluster(
+              _pinotHelixResourceManager.getHelixClusterName()).build();
+      Map<String, String> properties = new TreeMap<>();
+      GroovyStaticAnalyzerConfig groovyConfig = GroovyStaticAnalyzerConfig.fromJson(body);
+      properties.put(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG,
+          groovyConfig == null ? null : groovyConfig.toJson());
+      admin.setConfig(configScope, properties);
+      GroovyFunctionEvaluator.setConfig(groovyConfig);
+      return new SuccessResponse("Updated Groovy Static Analyzer config.");
+    } catch (IOException e) {
+      throw new ControllerApplicationException(LOGGER, "Error converting request to cluster config",
+          Response.Status.BAD_REQUEST, e);
+    } catch (Exception e) {
+      throw new ControllerApplicationException(LOGGER, "Failed to update Groovy Static Analyzer config",
+          Response.Status.INTERNAL_SERVER_ERROR, e);
+    }
+  }
+
+  @GET
+  @Path("/cluster/configs/groovy/staticAnalyzerConfig/default")
+  @Authorize(targetType = TargetType.CLUSTER, action = Actions.Cluster.GET_GROOVY_STATIC_ANALYZER_CONFIG)
+  @Produces(MediaType.APPLICATION_JSON)
+  @ApiOperation(value = "Get the default configuration for Groovy Static analysis",
+      notes = "Get the default configuration for Groovy static analysis")
+  @ApiResponses(value = {
+      @ApiResponse(code = 200, message = "Success"),
+      @ApiResponse(code = 500, message = "Internal server error")
+  })
+  public GroovyStaticAnalyzerConfig getDefaultGroovyStaticAnalysisConfig() {
+    return GroovyStaticAnalyzerConfig.createDefault();
+  }
 }
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java b/pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java
index 96e4f27790a7..2fa066e991fb 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/auth/Actions.java
@@ -99,6 +99,8 @@ public static class Cluster {
     public static final String UPDATE_INSTANCE_PARTITIONS = "UpdateInstancePartitions";
     public static final String GET_RESPONSE_STORE = "GetResponseStore";
     public static final String DELETE_RESPONSE_STORE = "DeleteResponseStore";
+    public static final String GET_GROOVY_STATIC_ANALYZER_CONFIG = "GetGroovyStaticAnalyzerConfig";
+    public static final String UPDATE_GROOVY_STATIC_ANALYZER_CONFIG = "UpdateGroovyStaticAnalyzerConfig";
   }
 
   // Action names for table
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyFunctionEvaluatorTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyFunctionEvaluatorTest.java
index 29e28f475ec2..1f57dd193b99 100644
--- a/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyFunctionEvaluatorTest.java
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyFunctionEvaluatorTest.java
@@ -18,32 +18,123 @@
  */
 package org.apache.pinot.core.data.function;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import org.apache.pinot.segment.local.function.GroovyFunctionEvaluator;
+import org.apache.pinot.segment.local.function.GroovyStaticAnalyzerConfig;
 import org.apache.pinot.spi.data.readers.GenericRow;
-import org.testng.Assert;
 import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 import org.testng.collections.Lists;
 
+import static org.apache.pinot.segment.local.function.GroovyStaticAnalyzerConfig.getDefaultAllowedImports;
+import static org.apache.pinot.segment.local.function.GroovyStaticAnalyzerConfig.getDefaultAllowedReceivers;
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.fail;
+
 
 /**
  * Tests Groovy functions for transforming schema columns
  */
 public class GroovyFunctionEvaluatorTest {
+  @Test
+  public void testLegalGroovyScripts()
+      throws JsonProcessingException {
+    // TODO: Add separate tests for these rules: receivers, imports, static imports, and method names.
+    List<String> scripts = List.of(
+        "Groovy({2})",
+        "Groovy({![\"pinot_minion_totalOutputSegmentSize_Value\"].contains(\"\");2})",
+        "Groovy({airtime == null ? (arrdelay == null ? 0 : arrdelay.value) : airtime.value; 2}, airtime, arrdelay)"
+    );
+
+    GroovyStaticAnalyzerConfig config = new GroovyStaticAnalyzerConfig(
+        getDefaultAllowedReceivers(),
+        getDefaultAllowedImports(),
+        getDefaultAllowedImports(),
+        List.of("invoke", "execute"),
+        false);
+    GroovyFunctionEvaluator.setConfig(config);
+
+    for (String script : scripts) {
+      GroovyFunctionEvaluator groovyFunctionEvaluator = new GroovyFunctionEvaluator(script);
+      GenericRow row = new GenericRow();
+      Object result = groovyFunctionEvaluator.evaluate(row);
+      assertEquals(2, result);
+    }
+  }
+
+  @Test
+  public void testIllegalGroovyScripts()
+      throws JsonProcessingException {
+    // TODO: Add separate tests for these rules: receivers, imports, static imports, and method names.
+    List<String> scripts = List.of(
+        "Groovy({\"ls\".execute()})",
+        "Groovy({[\"ls\"].execute()})",
+        "Groovy({System.exit(5)})",
+        "Groovy({System.metaClass.methods.each { method -> if (method.name.md5() == "
+            + "\"f24f62eeb789199b9b2e467df3b1876b\") {method.invoke(System, 10)} }})",
+        "Groovy({System.metaClass.methods.each { method -> if (method.name.reverse() == (\"ti\" + \"xe\")) "
+            + "{method.invoke(System, 10)} }})",
+        "groovy({def args = [\"QuickStart\", \"-type\", \"REALTIME\"] as String[]; "
+            + "org.apache.pinot.tools.admin.PinotAdministrator.main(args); 2})",
+        "Groovy({return [\"bash\", \"-c\", \"env\"].execute().text})"
+    );
+
+    GroovyStaticAnalyzerConfig config = new GroovyStaticAnalyzerConfig(
+        getDefaultAllowedReceivers(),
+        getDefaultAllowedImports(),
+        getDefaultAllowedImports(),
+        List.of("invoke", "execute"),
+        false);
+    GroovyFunctionEvaluator.setConfig(config);
+
+    for (String script : scripts) {
+      try {
+        new GroovyFunctionEvaluator(script);
+        fail("Groovy analyzer failed to catch malicious script");
+      } catch (Exception ignored) {
+      }
+    }
+  }
+
+  @Test
+  public void testUpdatingConfiguration()
+      throws JsonProcessingException {
+    // TODO: Figure out how to test this with the singleton initializer
+    // These tests would pass by default but the configuration will be updated so that they fail
+    List<String> scripts = List.of(
+        "Groovy({2})",
+        "Groovy({![\"pinot_minion_totalOutputSegmentSize_Value\"].contains(\"\");2})",
+        "Groovy({airtime == null ? (arrdelay == null ? 0 : arrdelay.value) : airtime.value; 2}, airtime, arrdelay)"
+    );
+
+    GroovyStaticAnalyzerConfig config =
+        new GroovyStaticAnalyzerConfig(List.of(), List.of(), List.of(), List.of(), false);
+    GroovyFunctionEvaluator.setConfig(config);
+
+    for (String script : scripts) {
+      try {
+        GroovyFunctionEvaluator groovyFunctionEvaluator = new GroovyFunctionEvaluator(script);
+        GenericRow row = new GenericRow();
+        groovyFunctionEvaluator.evaluate(row);
+        fail(String.format("Groovy analyzer failed to catch malicious script: %s", script));
+      } catch (Exception ignored) {
+      }
+    }
+  }
 
   @Test(dataProvider = "groovyFunctionEvaluationDataProvider")
   public void testGroovyFunctionEvaluation(String transformFunction, List<String> arguments, GenericRow genericRow,
       Object expectedResult) {
 
     GroovyFunctionEvaluator groovyExpressionEvaluator = new GroovyFunctionEvaluator(transformFunction);
-    Assert.assertEquals(groovyExpressionEvaluator.getArguments(), arguments);
+    assertEquals(groovyExpressionEvaluator.getArguments(), arguments);
 
     Object result = groovyExpressionEvaluator.evaluate(genericRow);
-    Assert.assertEquals(result, expectedResult);
+    assertEquals(result, expectedResult);
   }
 
   @DataProvider(name = "groovyFunctionEvaluationDataProvider")
@@ -108,20 +199,26 @@ public Object[][] groovyFunctionEvaluationDataProvider() {
     GenericRow genericRow9 = new GenericRow();
     genericRow9.putValue("ArrTime", 101);
     genericRow9.putValue("ArrTimeV2", null);
-    entries.add(new Object[]{"Groovy({ArrTimeV2 != null ? ArrTimeV2: ArrTime }, ArrTime, ArrTimeV2)",
-        Lists.newArrayList("ArrTime", "ArrTimeV2"), genericRow9, 101});
+    entries.add(new Object[]{
+        "Groovy({ArrTimeV2 != null ? ArrTimeV2: ArrTime }, ArrTime, ArrTimeV2)",
+        Lists.newArrayList("ArrTime", "ArrTimeV2"), genericRow9, 101
+    });
 
     GenericRow genericRow10 = new GenericRow();
     String jello = "Jello";
     genericRow10.putValue("jello", jello);
-    entries.add(new Object[]{"Groovy({jello != null ? jello.length() : \"Jello\" }, jello)",
-        Lists.newArrayList("jello"), genericRow10, 5});
+    entries.add(new Object[]{
+        "Groovy({jello != null ? jello.length() : \"Jello\" }, jello)",
+        Lists.newArrayList("jello"), genericRow10, 5
+    });
 
     //Invalid groovy script
     GenericRow genericRow11 = new GenericRow();
     genericRow11.putValue("nullValue", null);
-    entries.add(new Object[]{"Groovy({nullValue == null ? nullValue.length() : \"Jello\" }, nullValue)",
-        Lists.newArrayList("nullValue"), genericRow11, null});
+    entries.add(new Object[]{
+        "Groovy({nullValue == null ? nullValue.length() : \"Jello\" }, nullValue)",
+        Lists.newArrayList("nullValue"), genericRow11, null
+    });
     return entries.toArray(new Object[entries.size()][]);
   }
 }
diff --git a/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyStaticAnalyzerConfigTest.java b/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyStaticAnalyzerConfigTest.java
new file mode 100644
index 000000000000..d0a0c1c04934
--- /dev/null
+++ b/pinot-core/src/test/java/org/apache/pinot/core/data/function/GroovyStaticAnalyzerConfigTest.java
@@ -0,0 +1,141 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.core.data.function;
+
+import java.util.Iterator;
+import java.util.List;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.segment.local.function.GroovyStaticAnalyzerConfig;
+import org.apache.pinot.spi.utils.JsonUtils;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+
+/**
+ * Test serialization and deserialization.
+ */
+public class GroovyStaticAnalyzerConfigTest {
+  @Test
+  public void testEmptyConfig()
+      throws Exception {
+    GroovyStaticAnalyzerConfig config = new GroovyStaticAnalyzerConfig(null, null, null, null, false);
+    String encodedConfig = JsonUtils.objectToString(config);
+    GroovyStaticAnalyzerConfig decodedConfig =
+        JsonUtils.stringToObject(encodedConfig, GroovyStaticAnalyzerConfig.class);
+
+    Assert.assertNull(decodedConfig.getAllowedReceivers());
+    Assert.assertNull(decodedConfig.getAllowedImports());
+    Assert.assertNull(decodedConfig.getAllowedStaticImports());
+    Assert.assertNull(decodedConfig.getDisallowedMethodNames());
+  }
+
+  @Test
+  public void testAllowedReceivers()
+      throws Exception {
+    GroovyStaticAnalyzerConfig config = new GroovyStaticAnalyzerConfig(
+        GroovyStaticAnalyzerConfig.getDefaultAllowedReceivers(), null, null, null, false);
+    String encodedConfig = JsonUtils.objectToString(config);
+    GroovyStaticAnalyzerConfig decodedConfig =
+        JsonUtils.stringToObject(encodedConfig, GroovyStaticAnalyzerConfig.class);
+
+    Assert.assertEquals(GroovyStaticAnalyzerConfig.getDefaultAllowedReceivers(), decodedConfig.getAllowedReceivers());
+    Assert.assertNull(decodedConfig.getAllowedImports());
+    Assert.assertNull(decodedConfig.getAllowedStaticImports());
+    Assert.assertNull(decodedConfig.getDisallowedMethodNames());
+  }
+
+  @Test
+  public void testAllowedImports()
+      throws Exception {
+    GroovyStaticAnalyzerConfig config =
+        new GroovyStaticAnalyzerConfig(null, GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), null, null, false);
+    String encodedConfig = JsonUtils.objectToString(config);
+    GroovyStaticAnalyzerConfig decodedConfig =
+        JsonUtils.stringToObject(encodedConfig, GroovyStaticAnalyzerConfig.class);
+
+    Assert.assertNull(decodedConfig.getAllowedReceivers());
+    Assert.assertEquals(GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), decodedConfig.getAllowedImports());
+    Assert.assertNull(decodedConfig.getAllowedStaticImports());
+    Assert.assertNull(decodedConfig.getDisallowedMethodNames());
+  }
+
+  @Test
+  public void testAllowedStaticImports()
+      throws Exception {
+    GroovyStaticAnalyzerConfig config =
+        new GroovyStaticAnalyzerConfig(null, null, GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), null, false);
+    String encodedConfig = JsonUtils.objectToString(config);
+    GroovyStaticAnalyzerConfig decodedConfig =
+        JsonUtils.stringToObject(encodedConfig, GroovyStaticAnalyzerConfig.class);
+
+    Assert.assertNull(decodedConfig.getAllowedReceivers());
+    Assert.assertNull(decodedConfig.getAllowedImports());
+    Assert.assertEquals(GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), decodedConfig.getAllowedStaticImports());
+    Assert.assertNull(decodedConfig.getDisallowedMethodNames());
+    Assert.assertFalse(decodedConfig.isMethodDefinitionAllowed());
+  }
+
+  @Test
+  public void testDisallowedMethodNames()
+      throws Exception {
+    GroovyStaticAnalyzerConfig config =
+        new GroovyStaticAnalyzerConfig(null, null, null, List.of("method1", "method2"), false);
+    String encodedConfig = JsonUtils.objectToString(config);
+    GroovyStaticAnalyzerConfig decodedConfig =
+        JsonUtils.stringToObject(encodedConfig, GroovyStaticAnalyzerConfig.class);
+
+    Assert.assertNull(decodedConfig.getAllowedReceivers());
+    Assert.assertNull(decodedConfig.getAllowedImports());
+    Assert.assertNull(decodedConfig.getAllowedStaticImports());
+    Assert.assertEquals(List.of("method1", "method2"), decodedConfig.getDisallowedMethodNames());
+  }
+
+  @Test(dataProvider = "config_provider")
+  public void testToZnRecord(GroovyStaticAnalyzerConfig config)
+      throws Exception {
+    ZNRecord zr = config.toZNRecord();
+    GroovyStaticAnalyzerConfig znConfig = GroovyStaticAnalyzerConfig.fromZNRecord(zr);
+    Assert.assertTrue(equals(znConfig, config));
+  }
+
+  @DataProvider(name = "config_provider")
+  Iterator<GroovyStaticAnalyzerConfig> configProvider() {
+    return List.of(
+        new GroovyStaticAnalyzerConfig(null, null, null, List.of("method1", "method2"), false),
+        new GroovyStaticAnalyzerConfig(
+            GroovyStaticAnalyzerConfig.getDefaultAllowedReceivers(), null, null, null, false),
+        new GroovyStaticAnalyzerConfig(null, GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), null, null, false),
+        new GroovyStaticAnalyzerConfig(null, null, GroovyStaticAnalyzerConfig.getDefaultAllowedImports(), null, false),
+        new GroovyStaticAnalyzerConfig(null, null, null, List.of("method1", "method2"), false)
+    ).iterator();
+  }
+
+  private boolean equals(GroovyStaticAnalyzerConfig a, GroovyStaticAnalyzerConfig b) {
+    return a != null && b != null
+        && (a.getAllowedStaticImports() == b.getAllowedStaticImports()
+        || a.getAllowedStaticImports().equals(b.getAllowedStaticImports()))
+        && (a.getAllowedImports() == null && b.getAllowedImports() == null
+        || a.getAllowedImports().equals(b.getAllowedImports()))
+        && (a.getAllowedReceivers() == null && b.getAllowedReceivers() == null
+        || a.getAllowedReceivers().equals(b.getAllowedReceivers()))
+        && (a.getDisallowedMethodNames() == null && b.getDisallowedMethodNames() == null
+        || a.getDisallowedMethodNames().equals(b.getDisallowedMethodNames()));
+  }
+}
diff --git a/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
index 8dab73386fed..b88eda178204 100644
--- a/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
+++ b/pinot-minion/src/main/java/org/apache/pinot/minion/BaseMinionStarter.java
@@ -61,6 +61,7 @@
 import org.apache.pinot.minion.executor.PinotTaskExecutorFactory;
 import org.apache.pinot.minion.executor.TaskExecutorFactoryRegistry;
 import org.apache.pinot.minion.taskfactory.TaskFactoryRegistry;
+import org.apache.pinot.segment.local.function.GroovyFunctionEvaluator;
 import org.apache.pinot.spi.crypt.PinotCrypterFactory;
 import org.apache.pinot.spi.env.PinotConfiguration;
 import org.apache.pinot.spi.filesystem.PinotFSFactory;
@@ -272,6 +273,10 @@ public void start()
       minionContext.setSSLContext(sslContext);
     }
 
+    // Initializing Groovy execution security
+    GroovyFunctionEvaluator.configureGroovySecurity(_config.getProperty(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG));
+    GroovyFunctionEvaluator.setMetrics(minionMetrics, MinionMeter.GROOVY_SECURITY_VIOLATIONS);
+
     // Join the Helix cluster
     LOGGER.info("Joining the Helix cluster");
     _helixManager.getStateMachineEngine().registerStateModelFactory("Task", new TaskStateModelFactory(_helixManager,
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyFunctionEvaluator.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyFunctionEvaluator.java
index 52f36465bb16..d324b8949bc9 100644
--- a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyFunctionEvaluator.java
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyFunctionEvaluator.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pinot.segment.local.function;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import groovy.lang.Binding;
@@ -27,7 +28,14 @@
 import java.util.List;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
+import org.apache.pinot.common.metrics.AbstractMetrics;
 import org.apache.pinot.spi.data.readers.GenericRow;
+import org.codehaus.groovy.ast.expr.MethodCallExpression;
+import org.codehaus.groovy.control.CompilerConfiguration;
+import org.codehaus.groovy.control.customizers.ImportCustomizer;
+import org.codehaus.groovy.control.customizers.SecureASTCustomizer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 
 /**
@@ -45,6 +53,7 @@
  *  ]
  */
 public class GroovyFunctionEvaluator implements FunctionEvaluator {
+  private static final Logger LOGGER = LoggerFactory.getLogger(GroovyFunctionEvaluator.class);
 
   private static final String GROOVY_EXPRESSION_PREFIX = "Groovy";
   private static final String GROOVY_FUNCTION_REGEX = "Groovy\\(\\{(?<script>.+)}(,(?<arguments>.+))?\\)";
@@ -53,6 +62,9 @@ public class GroovyFunctionEvaluator implements FunctionEvaluator {
   private static final String ARGUMENTS_GROUP_NAME = "arguments";
   private static final String SCRIPT_GROUP_NAME = "script";
   private static final String ARGUMENTS_SEPARATOR = ",";
+  private static GroovyStaticAnalyzerConfig _config;
+  private static AbstractMetrics _metrics;
+  private static AbstractMetrics.Meter _securityViolationCounter;
 
   private final List<String> _arguments;
   private final int _numArguments;
@@ -72,13 +84,65 @@ public GroovyFunctionEvaluator(String closure) {
     }
     _numArguments = _arguments.size();
     _binding = new Binding();
-    _script = new GroovyShell(_binding).parse(matcher.group(SCRIPT_GROUP_NAME));
+    final String scriptText = matcher.group(SCRIPT_GROUP_NAME);
+
+    final GroovyStaticAnalyzerConfig groovyStaticAnalyzerConfig = getConfig();
+    _script = createSafeShell(_binding, groovyStaticAnalyzerConfig).parse(scriptText);
   }
 
   public static String getGroovyExpressionPrefix() {
     return GROOVY_EXPRESSION_PREFIX;
   }
 
+  /**
+   * This will create a Groovy Shell that is configured with static syntax analysis. This static syntax analysis
+   * will that any script which is run is restricted to a specific list of allowed operations, thus making it harder
+   * to execute malicious code.
+   *
+   * @param binding Binding instance to be used by Groovy Shell.
+   * @param groovyConfig GroovyStaticAnalyzerConfig instance to be used for static syntax analysis.
+   * @return GroovyShell instance with static syntax analysis.
+   */
+  private GroovyShell createSafeShell(Binding binding, GroovyStaticAnalyzerConfig groovyConfig) {
+    CompilerConfiguration config = new CompilerConfiguration();
+
+    if (groovyConfig != null) {
+      ImportCustomizer imports = new ImportCustomizer().addStaticStars("java.lang.Math");
+      SecureASTCustomizer secure = new SecureASTCustomizer();
+
+      secure.addExpressionCheckers(expression -> {
+        if (expression instanceof MethodCallExpression) {
+          MethodCallExpression method = (MethodCallExpression) expression;
+          return !groovyConfig.getDisallowedMethodNames().contains(method.getMethodAsString());
+        } else {
+          return true;
+        }
+      });
+
+      secure.setConstantTypesClassesWhiteList(GroovyStaticAnalyzerConfig.getDefaultAllowedTypes());
+      secure.setImportsWhitelist(groovyConfig.getAllowedImports());
+      secure.setStaticImportsWhitelist(groovyConfig.getAllowedImports());
+      secure.setReceiversWhiteList(groovyConfig.getAllowedReceivers());
+
+      // Block all * imports
+      secure.setStaticStarImportsWhitelist(groovyConfig.getAllowedImports());
+      secure.setStarImportsWhitelist(groovyConfig.getAllowedImports());
+
+      // Allow all expression and token types
+      secure.setExpressionsBlacklist(List.of());
+      secure.setTokensBlacklist(List.of());
+
+      secure.setMethodDefinitionAllowed(groovyConfig.isMethodDefinitionAllowed());
+      secure.setIndirectImportCheckEnabled(true);
+      secure.setClosuresAllowed(true);
+      secure.setPackageAllowed(false);
+
+      config.addCompilationCustomizers(imports, secure);
+    }
+
+    return new GroovyShell(binding, config);
+  }
+
   @Override
   public List<String> getArguments() {
     return _arguments;
@@ -96,6 +160,10 @@ public Object evaluate(GenericRow genericRow) {
     }
     try {
       return _script.run();
+    } catch (SecurityException ex) {
+      LOGGER.error("An insecure Groovy script execution caught: {}", _expression);
+      incrementSecurityViolationCounter();
+      throw ex;
     } catch (Exception e) {
       if (hasNullArgument) {
         return null;
@@ -110,11 +178,74 @@ public Object evaluate(Object[] values) {
     for (int i = 0; i < _numArguments; i++) {
       _binding.setVariable(_arguments.get(i), values[i]);
     }
-    return _script.run();
+    try {
+      return _script.run();
+    } catch (SecurityException ex) {
+      LOGGER.error("An insecure Groovy script execution caught: {}", _expression);
+      incrementSecurityViolationCounter();
+      throw ex;
+    }
   }
 
   @Override
   public String toString() {
     return _expression;
   }
+
+  private static GroovyStaticAnalyzerConfig getConfig() {
+    synchronized (GroovyFunctionEvaluator.class) {
+      return _config;
+    }
+  }
+
+  private static void incrementSecurityViolationCounter() {
+    synchronized (GroovyFunctionEvaluator.class) {
+      if (_metrics != null) {
+        _metrics.addMeteredGlobalValue(_securityViolationCounter, 1);
+      }
+    }
+  }
+
+  public static void setMetrics(AbstractMetrics metrics, AbstractMetrics.Meter securityViolationCounter) {
+    synchronized (GroovyFunctionEvaluator.class) {
+      _metrics = metrics;
+      _securityViolationCounter = securityViolationCounter;
+    }
+  }
+
+  public static void configureGroovySecurity(String groovyASTConfig) throws Exception {
+    try {
+      if (groovyASTConfig != null) {
+        setConfig(GroovyStaticAnalyzerConfig.fromJson(groovyASTConfig));
+      } else {
+        LOGGER.info("No Groovy Security Configuration found, Groovy static analysis is disabled");
+      }
+    } catch (Exception ex) {
+      // If there was an issue reading the security configuration then send an exception to the startup routine.
+      LOGGER.error("Failed to read config from ZK");
+      throw ex;
+    }
+  }
+
+
+  /**
+   * Initialize or update the configuration for the Groovy Static Analyzer.
+   * @param config GroovyStaticAnalyzerConfig instance to be used for static syntax analysis.
+   */
+  public static void setConfig(GroovyStaticAnalyzerConfig config)
+      throws JsonProcessingException {
+    synchronized (GroovyFunctionEvaluator.class) {
+      if (_config == null) {
+        LOGGER.info("Initializing Groovy Static Analyzer: {}", config.toJson());
+        _config = config;
+      } else {
+        // It's fine to update the configuration because the static variable stores the address to a configuration
+        // object and whenever a GroovyFunctionEvaluator is created it makes a local copy of that address by reading
+        // the static variable atomically.  So, if the static config variable changes it will have no effect on
+        // any currently running evaluators.
+        LOGGER.info("Updating Groovy Static Analyzer: {}", config.toJson());
+        _config = config;
+      }
+    }
+  }
 }
diff --git a/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyStaticAnalyzerConfig.java b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyStaticAnalyzerConfig.java
new file mode 100644
index 000000000000..6e3c84a1f560
--- /dev/null
+++ b/pinot-segment-local/src/main/java/org/apache/pinot/segment/local/function/GroovyStaticAnalyzerConfig.java
@@ -0,0 +1,153 @@
+/**
+ * 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.
+ */
+package org.apache.pinot.segment.local.function;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Preconditions;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.List;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.helix.zookeeper.datamodel.ZNRecord;
+import org.apache.pinot.spi.utils.JsonUtils;
+
+
+public class GroovyStaticAnalyzerConfig {
+  private final List<String> _allowedReceivers;
+  private final List<String> _allowedImports;
+  private final List<String> _allowedStaticImports;
+  private final List<String> _disallowedMethodNames;
+  private final boolean _methodDefinitionAllowed;
+
+  public GroovyStaticAnalyzerConfig(
+      @JsonProperty("allowedReceivers")
+      List<String> allowedReceivers,
+      @JsonProperty("allowedImports")
+      List<String> allowedImports,
+      @JsonProperty("allowedStaticImports")
+      List<String> allowedStaticImports,
+      @JsonProperty("disallowedMethodNames")
+      List<String> disallowedMethodNames,
+      @JsonProperty("methodDefinitionAllowed")
+      boolean methodDefinitionAllowed) {
+    _allowedImports = allowedImports;
+    _allowedReceivers = allowedReceivers;
+    _allowedStaticImports = allowedStaticImports;
+    _disallowedMethodNames = disallowedMethodNames;
+    _methodDefinitionAllowed = methodDefinitionAllowed;
+  }
+
+  @JsonProperty("allowedReceivers")
+  public List<String> getAllowedReceivers() {
+    return _allowedReceivers;
+  }
+
+  @JsonProperty("allowedImports")
+  public List<String> getAllowedImports() {
+    return _allowedImports;
+  }
+
+  @JsonProperty("allowedStaticImports")
+  public List<String> getAllowedStaticImports() {
+    return _allowedStaticImports;
+  }
+
+  @JsonProperty("disallowedMethodNames")
+  public List<String> getDisallowedMethodNames() {
+    return _disallowedMethodNames;
+  }
+
+  @JsonProperty("methodDefinitionAllowed")
+  public boolean isMethodDefinitionAllowed() {
+    return _methodDefinitionAllowed;
+  }
+
+  public ZNRecord toZNRecord() throws JsonProcessingException {
+    ZNRecord record = new ZNRecord("groovySecurityConfiguration");
+    record.setSimpleField("staticAnalyzerConfig", toJson());
+    return record;
+  }
+
+  public static GroovyStaticAnalyzerConfig fromZNRecord(ZNRecord zr) throws JsonProcessingException {
+    Preconditions.checkArgument(zr.getId().equals("groovySecurityConfiguration"),
+        "Expected ZNRecord with ID \"groovySecurityConfiguration\" but got {}", zr.getId());
+
+    final String configJson = zr.getSimpleField("staticAnalyzerConfig");
+    return fromJson(configJson);
+  }
+
+  public String toJson() throws JsonProcessingException {
+    return JsonUtils.objectToString(this);
+  }
+
+  public static GroovyStaticAnalyzerConfig fromJson(String configJson) throws JsonProcessingException {
+    Preconditions.checkState(StringUtils.isNotBlank(configJson), "Empty groovySecurityConfiguration JSON string");
+
+    return JsonUtils.stringToObject(configJson, GroovyStaticAnalyzerConfig.class);
+  }
+
+  public static List<Class> getDefaultAllowedTypes() {
+    return List.of(
+        Integer.class,
+        Float.class,
+        Long.class,
+        Double.class,
+        String.class,
+        Object.class,
+        Byte.class,
+        BigDecimal.class,
+        BigInteger.class,
+        Integer.TYPE,
+        Long.TYPE,
+        Float.TYPE,
+        Double.TYPE,
+        Byte.TYPE
+    );
+  }
+
+  public static List<String> getDefaultAllowedReceivers() {
+    return List.of(
+        String.class.getName(),
+        Math.class.getName(),
+        java.util.List.class.getName(),
+        Object.class.getName(),
+        java.util.Map.class.getName()
+    );
+  }
+
+  public static List<String> getDefaultAllowedImports() {
+    return List.of(
+        Math.class.getName(),
+        java.util.List.class.getName(),
+        String.class.getName(),
+        java.util.Map.class.getName()
+    );
+  }
+
+  public static GroovyStaticAnalyzerConfig createDefault() {
+    return new GroovyStaticAnalyzerConfig(
+        GroovyStaticAnalyzerConfig.getDefaultAllowedReceivers(),
+        GroovyStaticAnalyzerConfig.getDefaultAllowedImports(),
+        GroovyStaticAnalyzerConfig.getDefaultAllowedImports(),
+        List.of("execute", "invoke"),
+        false
+    );
+  }
+}
diff --git a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
index e75c38fe0b5b..bb5fc6b12825 100644
--- a/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
+++ b/pinot-server/src/main/java/org/apache/pinot/server/starter/helix/BaseServerStarter.java
@@ -78,6 +78,7 @@
 import org.apache.pinot.core.query.scheduler.resources.ResourceManager;
 import org.apache.pinot.core.transport.ListenerConfig;
 import org.apache.pinot.core.util.ListenerConfigUtil;
+import org.apache.pinot.segment.local.function.GroovyFunctionEvaluator;
 import org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneIndexRefreshManager;
 import org.apache.pinot.segment.local.realtime.impl.invertedindex.RealtimeLuceneTextIndexSearcherPool;
 import org.apache.pinot.segment.local.utils.SegmentPreprocessThrottler;
@@ -671,6 +672,11 @@ public void start()
     _adminApiApplication = createServerAdminApp();
     _adminApiApplication.start(_listenerConfigs);
 
+    // Initializing Groovy execution security
+    GroovyFunctionEvaluator.configureGroovySecurity(
+        _serverConf.getProperty(CommonConstants.GROOVY_STATIC_ANALYZER_CONFIG));
+    GroovyFunctionEvaluator.setMetrics(serverMetrics, ServerMeter.GROOVY_SECURITY_VIOLATIONS);
+
     // Init QueryRewriterFactory
     LOGGER.info("Initializing QueryRewriterFactory");
     QueryRewriterFactory.init(_serverConf.getProperty(Server.CONFIG_OF_SERVER_QUERY_REWRITER_CLASS_NAMES));
diff --git a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
index d4dbe98eaade..4dc11eb69c67 100644
--- a/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
+++ b/pinot-spi/src/main/java/org/apache/pinot/spi/utils/CommonConstants.java
@@ -29,6 +29,8 @@
 
 
 public class CommonConstants {
+  public static final String GROOVY_STATIC_ANALYZER_CONFIG = "pinot.server.groovy.static.analyzer";
+
   private CommonConstants() {
   }
 
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java b/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
index d5033b4e907f..e526f9f17616 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
@@ -50,6 +50,7 @@ public static void main(String[] args)
   protected Map<String, Object> getConfigOverrides() {
     Map<String, Object> configOverrides = new HashMap<>();
     configOverrides.put(CommonConstants.Server.CONFIG_OF_ENABLE_THREAD_CPU_TIME_MEASUREMENT, true);
+    configOverrides.put(CommonConstants.Broker.DISABLE_GROOVY, false);
     return configOverrides;
   }