diff --git a/web-console/src/dialogs/compaction-dynamic-config-dialog/compaction-dynamic-config-dialog.tsx b/web-console/src/dialogs/compaction-dynamic-config-dialog/compaction-dynamic-config-dialog.tsx
index 1b6bb587aab2..0e352784f625 100644
--- a/web-console/src/dialogs/compaction-dynamic-config-dialog/compaction-dynamic-config-dialog.tsx
+++ b/web-console/src/dialogs/compaction-dynamic-config-dialog/compaction-dynamic-config-dialog.tsx
@@ -20,35 +20,18 @@ import { Button, Classes, Code, Dialog, Intent } from '@blueprintjs/core';
 import { IconNames } from '@blueprintjs/icons';
 import React, { useState } from 'react';
 
-import type { Field } from '../../components';
-import { AutoForm, ExternalLink, Loader } from '../../components';
+import type { FormJsonTabs } from '../../components';
+import { AutoForm, ExternalLink, FormJsonSelector, JsonInput, Loader } from '../../components';
+import type { CompactionDynamicConfig } from '../../druid-models';
+import {
+  COMPACTION_DYNAMIC_CONFIG_DEFAULT_MAX,
+  COMPACTION_DYNAMIC_CONFIG_DEFAULT_RATIO,
+  COMPACTION_DYNAMIC_CONFIG_FIELDS,
+} from '../../druid-models';
 import { useQueryManager } from '../../hooks';
 import { getLink } from '../../links';
 import { Api, AppToaster } from '../../singletons';
-import { getDruidErrorMessage } from '../../utils';
-
-interface CompactionDynamicConfig {
-  compactionTaskSlotRatio: number;
-  maxCompactionTaskSlots: number;
-}
-
-const DEFAULT_RATIO = 0.1;
-const DEFAULT_MAX = 2147483647;
-const COMPACTION_DYNAMIC_CONFIG_FIELDS: Field<CompactionDynamicConfig>[] = [
-  {
-    name: 'compactionTaskSlotRatio',
-    type: 'ratio',
-    defaultValue: DEFAULT_RATIO,
-    info: <>The ratio of the total task slots to the compaction task slots.</>,
-  },
-  {
-    name: 'maxCompactionTaskSlots',
-    type: 'number',
-    defaultValue: DEFAULT_MAX,
-    info: <>The maximum number of task slots for compaction tasks</>,
-    min: 0,
-  },
-];
+import { getDruidErrorMessage, wait } from '../../utils';
 
 export interface CompactionDynamicConfigDialogProps {
   onClose(): void;
@@ -58,21 +41,20 @@ export const CompactionDynamicConfigDialog = React.memo(function CompactionDynam
   props: CompactionDynamicConfigDialogProps,
 ) {
   const { onClose } = props;
+  const [currentTab, setCurrentTab] = useState<FormJsonTabs>('form');
   const [dynamicConfig, setDynamicConfig] = useState<
     Partial<CompactionDynamicConfig> | undefined
   >();
+  const [jsonError, setJsonError] = useState<Error | undefined>();
 
   useQueryManager<null, Record<string, any>>({
     initQuery: null,
     processQuery: async (_, cancelToken) => {
       try {
-        const c = (
-          await Api.instance.get('/druid/coordinator/v1/config/compaction', { cancelToken })
-        ).data;
-        setDynamicConfig({
-          compactionTaskSlotRatio: c.compactionTaskSlotRatio ?? DEFAULT_RATIO,
-          maxCompactionTaskSlots: c.maxCompactionTaskSlots ?? DEFAULT_MAX,
+        const configResp = await Api.instance.get('/druid/indexer/v1/compaction/config/cluster', {
+          cancelToken,
         });
+        setDynamicConfig(configResp.data || {});
       } catch (e) {
         AppToaster.show({
           icon: IconNames.ERROR,
@@ -88,26 +70,28 @@ export const CompactionDynamicConfigDialog = React.memo(function CompactionDynam
   async function saveConfig() {
     if (!dynamicConfig) return;
     try {
-      // This API is terrible. https://druid.apache.org/docs/latest/operations/api-reference#automatic-compaction-configuration
-      await Api.instance.post(
-        `/druid/coordinator/v1/config/compaction/taskslots?ratio=${
-          dynamicConfig.compactionTaskSlotRatio ?? DEFAULT_RATIO
-        }&max=${dynamicConfig.maxCompactionTaskSlots ?? DEFAULT_MAX}`,
-        {},
-      );
+      await Api.instance.post('/druid/indexer/v1/compaction/config/cluster', dynamicConfig);
     } catch (e) {
       AppToaster.show({
         icon: IconNames.ERROR,
         intent: Intent.DANGER,
         message: `Could not save compaction dynamic config: ${getDruidErrorMessage(e)}`,
       });
+      return;
     }
 
     AppToaster.show({
       message: 'Saved compaction dynamic config',
       intent: Intent.SUCCESS,
     });
+
     onClose();
+
+    // Reload the page also because the datasources page pulls from different APIs depending on the setting of supervisor based compaction
+    if (location.hash.includes('#datasources')) {
+      await wait(1000); // Wait for a second to give the user time to read the toast
+      location.reload();
+    }
   }
 
   return (
@@ -135,17 +119,33 @@ export const CompactionDynamicConfigDialog = React.memo(function CompactionDynam
             <p>
               The maximum number of task slots used for compaction will be{' '}
               <Code>{`clamp(floor(${
-                dynamicConfig.compactionTaskSlotRatio ?? DEFAULT_RATIO
-              } * total_task_slots), 1, ${
-                dynamicConfig.maxCompactionTaskSlots ?? DEFAULT_MAX
+                dynamicConfig.compactionTaskSlotRatio ?? COMPACTION_DYNAMIC_CONFIG_DEFAULT_RATIO
+              } * total_task_slots), ${dynamicConfig.engine === 'msq' ? 2 : 1}, ${
+                dynamicConfig.maxCompactionTaskSlots ?? COMPACTION_DYNAMIC_CONFIG_DEFAULT_MAX
               })`}</Code>
               .
             </p>
-            <AutoForm
-              fields={COMPACTION_DYNAMIC_CONFIG_FIELDS}
-              model={dynamicConfig}
-              onChange={setDynamicConfig}
+            <FormJsonSelector
+              tab={currentTab}
+              onChange={t => {
+                setJsonError(undefined);
+                setCurrentTab(t);
+              }}
             />
+            {currentTab === 'form' ? (
+              <AutoForm
+                fields={COMPACTION_DYNAMIC_CONFIG_FIELDS}
+                model={dynamicConfig}
+                onChange={setDynamicConfig}
+              />
+            ) : (
+              <JsonInput
+                value={dynamicConfig}
+                height="50vh"
+                onChange={setDynamicConfig}
+                setError={setJsonError}
+              />
+            )}
           </div>
           <div className={Classes.DIALOG_FOOTER}>
             <div className={Classes.DIALOG_FOOTER_ACTIONS}>
@@ -154,6 +154,7 @@ export const CompactionDynamicConfigDialog = React.memo(function CompactionDynam
                 onClick={() => void saveConfig()}
                 intent={Intent.PRIMARY}
                 rightIcon={IconNames.TICK}
+                disabled={Boolean(jsonError)}
               />
             </div>
           </div>
diff --git a/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.scss b/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.scss
index 620e7f6ff227..e6da0a9f3431 100644
--- a/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.scss
+++ b/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.scss
@@ -36,11 +36,7 @@
     .global-info {
       position: absolute;
       bottom: 10px;
-      left: 30px;
-      right: 10px;
-      width: auto;
-      white-space: pre;
-      background: $gray1;
+      right: 20px;
     }
   }
 
diff --git a/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.tsx b/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.tsx
index 8ee0ffdbfbc3..b77fc3d3dda2 100644
--- a/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.tsx
+++ b/web-console/src/dialogs/compaction-history-dialog/compaction-history-dialog.tsx
@@ -16,15 +16,16 @@
  * limitations under the License.
  */
 
-import { Button, Callout, Classes, Dialog, Tab, Tabs, TabsExpander, Tag } from '@blueprintjs/core';
+import { Button, Classes, Dialog, Popover, Tab, Tabs, TabsExpander, Tag } from '@blueprintjs/core';
+import { IconNames } from '@blueprintjs/icons';
 import * as JSONBig from 'json-bigint-native';
 import React, { useState } from 'react';
 
-import { Loader, ShowValue } from '../../components';
+import { Loader, PopoverText, ShowValue } from '../../components';
 import type { CompactionConfig } from '../../druid-models';
 import { useQueryManager } from '../../hooks';
 import { Api } from '../../singletons';
-import { formatInteger, formatPercent, getApiArray } from '../../utils';
+import { formatInteger, formatPercent, getApiArrayFromKey } from '../../utils';
 import { DiffDialog } from '../diff-dialog/diff-dialog';
 
 import './compaction-history-dialog.scss';
@@ -46,7 +47,7 @@ function formatGlobalConfig(globalConfig: GlobalConfig): string {
   return [
     `compactionTaskSlotRatio: ${formatPercent(globalConfig.compactionTaskSlotRatio)}`,
     `maxCompactionTaskSlots: ${formatInteger(globalConfig.maxCompactionTaskSlots)}`,
-    `useAutoScaleSlots: ${globalConfig.useAutoScaleSlots}`,
+    `useAutoScaleSlots: ${Boolean(globalConfig.useAutoScaleSlots)}`,
   ].join('\n');
 }
 
@@ -65,8 +66,11 @@ export const CompactionHistoryDialog = React.memo(function CompactionHistoryDial
     initQuery: datasource,
     processQuery: async (datasource, cancelToken) => {
       try {
-        return await getApiArray<CompactionHistoryEntry>(
-          `/druid/coordinator/v1/config/compaction/${Api.encodePath(datasource)}/history?count=20`,
+        return await getApiArrayFromKey<CompactionHistoryEntry>(
+          `/druid/indexer/v1/compaction/config/datasources/${Api.encodePath(
+            datasource,
+          )}/history?count=20`,
+          'entries',
           cancelToken,
         );
       } catch (e) {
@@ -105,9 +109,16 @@ export const CompactionHistoryDialog = React.memo(function CompactionHistoryDial
                         downloadFilename={`compaction-history-${datasource}-version-${historyEntry.auditTime}.json`}
                       />
                       {historyEntry.globalConfig && (
-                        <Callout className="global-info">
-                          {formatGlobalConfig(historyEntry.globalConfig)}
-                        </Callout>
+                        <Popover
+                          className="global-info"
+                          content={
+                            <PopoverText>
+                              <pre>{formatGlobalConfig(historyEntry.globalConfig)}</pre>
+                            </PopoverText>
+                          }
+                        >
+                          <Button icon={IconNames.GLOBE} text="Global config" />
+                        </Popover>
                       )}
                     </>
                   }
diff --git a/web-console/src/dialogs/doctor-dialog/doctor-checks.tsx b/web-console/src/dialogs/doctor-dialog/doctor-checks.tsx
index 2104c046b884..17cd8acd1a46 100644
--- a/web-console/src/dialogs/doctor-dialog/doctor-checks.tsx
+++ b/web-console/src/dialogs/doctor-dialog/doctor-checks.tsx
@@ -16,6 +16,7 @@
  * limitations under the License.
  */
 
+import type { CompactionConfigs } from '../../druid-models';
 import { Api } from '../../singletons';
 import { deepGet, pluralIfNeeded, queryDruidSql } from '../../utils';
 import { postToSampler } from '../../utils/sampler';
@@ -373,10 +374,11 @@ ORDER BY "num_bad_time_chunks"`,
 
       if (sqlResult.length) {
         // Grab the auto-compaction definitions and ignore dataSources that already have auto-compaction
-        let compactionResult: any;
+        let compactionResult: CompactionConfigs;
         try {
-          compactionResult = (await Api.instance.get('/druid/coordinator/v1/config/compaction'))
-            .data;
+          compactionResult = (
+            await Api.instance.get('/druid/indexer/v1/compaction/config/datasources')
+          ).data;
         } catch (e) {
           controls.addIssue(`Could not get compaction config. Something is wrong.`);
           return;
diff --git a/web-console/src/druid-models/compaction-config/compaction-config.tsx b/web-console/src/druid-models/compaction-config/compaction-config.tsx
index 54f16adbfdcf..de4a2f47a843 100644
--- a/web-console/src/druid-models/compaction-config/compaction-config.tsx
+++ b/web-console/src/druid-models/compaction-config/compaction-config.tsx
@@ -31,6 +31,10 @@ export interface CompactionConfig {
   inputSegmentSizeBytes?: number;
 }
 
+export interface CompactionConfigs {
+  compactionConfigs: CompactionConfig[];
+}
+
 export const NOOP_INPUT_SEGMENT_SIZE_BYTES = 100000000000000;
 
 export function compactionConfigHasLegacyInputSegmentSizeBytesSet(
diff --git a/web-console/src/druid-models/compaction-dynamic-config/compaction-dynamic-config.tsx b/web-console/src/druid-models/compaction-dynamic-config/compaction-dynamic-config.tsx
new file mode 100644
index 000000000000..047f3f84f738
--- /dev/null
+++ b/web-console/src/druid-models/compaction-dynamic-config/compaction-dynamic-config.tsx
@@ -0,0 +1,97 @@
+/*
+ * 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.
+ */
+
+import { Code } from '@blueprintjs/core';
+
+import type { Field } from '../../components';
+import { deepGet } from '../../utils';
+
+export interface CompactionDynamicConfig {
+  compactionTaskSlotRatio: number;
+  maxCompactionTaskSlots: number;
+  compactionPolicy: { type: 'newestSegmentFirst'; priorityDatasource?: string | null };
+  useSupervisors: boolean;
+  engine: 'native' | 'msq';
+}
+
+export const COMPACTION_DYNAMIC_CONFIG_DEFAULT_RATIO = 0.1;
+export const COMPACTION_DYNAMIC_CONFIG_DEFAULT_MAX = 2147483647;
+export const COMPACTION_DYNAMIC_CONFIG_FIELDS: Field<CompactionDynamicConfig>[] = [
+  {
+    name: 'useSupervisors',
+    label: 'Use supervisors',
+    experimental: true,
+    type: 'boolean',
+    defaultValue: false,
+    info: (
+      <>
+        <p>
+          Whether compaction should be run on Overlord using supervisors instead of Coordinator
+          duties.
+        </p>
+        <p>Supervisor based compaction is an experimental feature.</p>
+      </>
+    ),
+  },
+  {
+    name: 'engine',
+    type: 'string',
+    defined: config => Boolean(config.useSupervisors),
+    defaultValue: 'native',
+    suggestions: ['native', 'msq'],
+    info: 'Engine to use for running compaction tasks, native or MSQ.',
+  },
+  {
+    name: 'compactionTaskSlotRatio',
+    type: 'ratio',
+    defaultValue: COMPACTION_DYNAMIC_CONFIG_DEFAULT_RATIO,
+    info: <>The ratio of the total task slots to the compaction task slots.</>,
+  },
+  {
+    name: 'maxCompactionTaskSlots',
+    type: 'number',
+    defaultValue: COMPACTION_DYNAMIC_CONFIG_DEFAULT_MAX,
+    info: <>The maximum number of task slots for compaction tasks</>,
+    min: 0,
+  },
+  {
+    name: 'compactionPolicy.type',
+    label: 'Compaction search policy',
+    type: 'string',
+    suggestions: ['newestSegmentFirst'],
+    info: (
+      <>
+        Currently, the only supported policy is <Code>newestSegmentFirst</Code>, which prioritizes
+        segments with more recent intervals for compaction.
+      </>
+    ),
+  },
+  {
+    name: 'compactionPolicy.priorityDatasource',
+    type: 'string',
+    defined: config => deepGet(config, 'compactionPolicy.type') === 'newestSegmentFirst',
+    placeholder: '(none)',
+    info: (
+      <>
+        Datasource to prioritize for compaction. The intervals of this datasource are chosen for
+        compaction before the intervals of any other datasource. Within this datasource, the
+        intervals are prioritized based on the chosen compaction policy.
+      </>
+    ),
+  },
+];
diff --git a/web-console/src/druid-models/coordinator-dynamic-config/coordinator-dynamic-config.tsx b/web-console/src/druid-models/coordinator-dynamic-config/coordinator-dynamic-config.tsx
index 4a7224dbee4d..3794ecb60376 100644
--- a/web-console/src/druid-models/coordinator-dynamic-config/coordinator-dynamic-config.tsx
+++ b/web-console/src/druid-models/coordinator-dynamic-config/coordinator-dynamic-config.tsx
@@ -157,26 +157,6 @@ export const COORDINATOR_DYNAMIC_CONFIG_FIELDS: Field<CoordinatorDynamicConfig>[
       </>
     ),
   },
-  {
-    name: 'turboLoadingNodes',
-    type: 'string-array',
-    experimental: true,
-    info: (
-      <>
-        <p>
-          List of Historical servers to place in turbo loading mode. These servers use a larger
-          thread-pool to load segments faster but at the cost of query performance. For servers
-          specified in <Code>turboLoadingNodes</Code>,{' '}
-          <Code>druid.coordinator.loadqueuepeon.http.batchSize</Code> is ignored and the coordinator
-          uses the value of the respective <Code>numLoadingThreads</Code> instead.
-        </p>
-        <p>
-          Please use this config with caution. All servers should eventually be removed from this
-          list once the segment loading on the respective historicals is finished.
-        </p>
-      </>
-    ),
-  },
   {
     name: 'killDataSourceWhitelist',
     label: 'Kill datasource whitelist',
@@ -265,4 +245,24 @@ export const COORDINATOR_DYNAMIC_CONFIG_FIELDS: Field<CoordinatorDynamicConfig>[
       </>
     ),
   },
+  {
+    name: 'turboLoadingNodes',
+    type: 'string-array',
+    experimental: true,
+    info: (
+      <>
+        <p>
+          List of Historical servers to place in turbo loading mode. These servers use a larger
+          thread-pool to load segments faster but at the cost of query performance. For servers
+          specified in <Code>turboLoadingNodes</Code>,{' '}
+          <Code>druid.coordinator.loadqueuepeon.http.batchSize</Code> is ignored and the coordinator
+          uses the value of the respective <Code>numLoadingThreads</Code> instead.
+        </p>
+        <p>
+          Please use this config with caution. All servers should eventually be removed from this
+          list once the segment loading on the respective historicals is finished.
+        </p>
+      </>
+    ),
+  },
 ];
diff --git a/web-console/src/druid-models/index.ts b/web-console/src/druid-models/index.ts
index e31eedeea738..2e23cf7ee2c2 100644
--- a/web-console/src/druid-models/index.ts
+++ b/web-console/src/druid-models/index.ts
@@ -18,6 +18,7 @@
 
 export * from './async-query/async-query';
 export * from './compaction-config/compaction-config';
+export * from './compaction-dynamic-config/compaction-dynamic-config';
 export * from './compaction-status/compaction-status';
 export * from './coordinator-dynamic-config/coordinator-dynamic-config';
 export * from './dart/dart-query-entry';
diff --git a/web-console/src/utils/druid-query.ts b/web-console/src/utils/druid-query.ts
index ee1ba1ef4a8c..d83cf5eb9b6a 100644
--- a/web-console/src/utils/druid-query.ts
+++ b/web-console/src/utils/druid-query.ts
@@ -364,6 +364,16 @@ export async function getApiArray<T = any>(url: string, cancelToken?: CancelToke
   return result;
 }
 
+export async function getApiArrayFromKey<T = any>(
+  url: string,
+  key: string,
+  cancelToken?: CancelToken,
+): Promise<T[]> {
+  const result = (await Api.instance.get(url, { cancelToken })).data?.[key];
+  if (!Array.isArray(result)) throw new Error('unexpected result');
+  return result;
+}
+
 export interface QueryExplanation {
   query: any;
   signature: { name: string; type: string }[];
diff --git a/web-console/src/views/datasources-view/datasources-view.tsx b/web-console/src/views/datasources-view/datasources-view.tsx
index ea0df40bb0a4..8ff4ec9f4239 100644
--- a/web-console/src/views/datasources-view/datasources-view.tsx
+++ b/web-console/src/views/datasources-view/datasources-view.tsx
@@ -48,6 +48,7 @@ import {
 import { DatasourceTableActionDialog } from '../../dialogs/datasource-table-action-dialog/datasource-table-action-dialog';
 import type {
   CompactionConfig,
+  CompactionConfigs,
   CompactionInfo,
   CompactionStatus,
   QueryWithContext,
@@ -614,17 +615,20 @@ GROUP BY 1, 2`;
           // Compaction
           auxiliaryQueries.push(async (datasourcesAndDefaultRules, cancelToken) => {
             try {
-              const compactionConfigsResp = await Api.instance.get<{
-                compactionConfigs: CompactionConfig[];
-              }>('/druid/coordinator/v1/config/compaction', { cancelToken });
+              const compactionConfigsAndMore = (
+                await Api.instance.get<CompactionConfigs>(
+                  '/druid/indexer/v1/compaction/config/datasources',
+                  { cancelToken },
+                )
+              ).data;
               const compactionConfigs = lookupBy(
-                compactionConfigsResp.data.compactionConfigs || [],
+                compactionConfigsAndMore.compactionConfigs || [],
                 c => c.dataSource,
               );
 
               const compactionStatusesResp = await Api.instance.get<{
                 latestStatus: CompactionStatus[];
-              }>('/druid/coordinator/v1/compaction/status', { cancelToken });
+              }>('/druid/indexer/v1/compaction/status/datasources', { cancelToken });
               const compactionStatuses = lookupBy(
                 compactionStatusesResp.data.latestStatus || [],
                 c => c.dataSource,
@@ -944,7 +948,12 @@ GROUP BY 1, 2`;
   private readonly saveCompaction = async (compactionConfig: CompactionConfig) => {
     if (!compactionConfig) return;
     try {
-      await Api.instance.post(`/druid/coordinator/v1/config/compaction`, compactionConfig);
+      await Api.instance.post(
+        `/druid/indexer/v1/compaction/config/datasources/${Api.encodePath(
+          compactionConfig.dataSource,
+        )}`,
+        compactionConfig,
+      );
       this.setState({ compactionDialogOpenOn: undefined });
       this.fetchData();
     } catch (e) {
@@ -968,7 +977,7 @@ GROUP BY 1, 2`;
         onClick: async () => {
           try {
             await Api.instance.delete(
-              `/druid/coordinator/v1/config/compaction/${Api.encodePath(datasource)}`,
+              `/druid/indexer/v1/compaction/config/datasources/${Api.encodePath(datasource)}`,
             );
             this.setState({ compactionDialogOpenOn: undefined }, () => this.fetchData());
           } catch (e) {