[8.10] [Fleet][Kafka] When compression is enabled, "None" shouldn't b…

…e an option for codec (#164416) (#164467)

# Backport

This will backport the following commits from `main` to `8.10`:
- [[Fleet][Kafka] When compression is enabled, "None" shouldn't be an
option for codec
(#164416)](#164416)

<!--- Backport version: 8.9.8 -->

### Questions ?
Please refer to the [Backport tool
documentation](https://github.com/sqren/backport)

<!--BACKPORT [{"author":{"name":"Konrad
Szwarc","email":"konrad.szwarc@elastic.co"},"sourceCommit":{"committedDate":"2023-08-22T14:33:39Z","message":"[Fleet][Kafka]
When compression is enabled, \"None\" shouldn't be an option for codec
(#164416)\n\nCloses
https://github.com/elastic/ingest-dev/issues/2327\r\n\r\n`None` option
removed from UI, tests aligned.
\r\n\r\n\r\n\r\nhttps://github.com/elastic/kibana/assets/29123534/b8b7ac7a-e3e9-4800-94b9-f80e1f42f044","sha":"c2a552ef66ac8e0c2a0372091ba85b3bf1fa7a3a","branchLabelMapping":{"^v8.11.0$":"main","^v(\\d+).(\\d+).\\d+$":"$1.$2"}},"sourcePullRequest":{"labels":["release_note:skip","Team:Fleet","v8.10.0","v8.11.0"],"number":164416,"url":"https://github.com/elastic/kibana/pull/164416","mergeCommit":{"message":"[Fleet][Kafka]
When compression is enabled, \"None\" shouldn't be an option for codec
(#164416)\n\nCloses
https://github.com/elastic/ingest-dev/issues/2327\r\n\r\n`None` option
removed from UI, tests aligned.
\r\n\r\n\r\n\r\nhttps://github.com/elastic/kibana/assets/29123534/b8b7ac7a-e3e9-4800-94b9-f80e1f42f044","sha":"c2a552ef66ac8e0c2a0372091ba85b3bf1fa7a3a"}},"sourceBranch":"main","suggestedTargetBranches":["8.10"],"targetPullRequestStates":[{"branch":"8.10","label":"v8.10.0","labelRegex":"^v(\\d+).(\\d+).\\d+$","isSourceBranch":false,"state":"NOT_CREATED"},{"branch":"main","label":"v8.11.0","labelRegex":"^v8.11.0$","isSourceBranch":true,"state":"MERGED","url":"https://github.com/elastic/kibana/pull/164416","number":164416,"mergeCommit":{"message":"[Fleet][Kafka]
When compression is enabled, \"None\" shouldn't be an option for codec
(#164416)\n\nCloses
https://github.com/elastic/ingest-dev/issues/2327\r\n\r\n`None` option
removed from UI, tests aligned.
\r\n\r\n\r\n\r\nhttps://github.com/elastic/kibana/assets/29123534/b8b7ac7a-e3e9-4800-94b9-f80e1f42f044","sha":"c2a552ef66ac8e0c2a0372091ba85b3bf1fa7a3a"}}]}]
BACKPORT-->

x-pack/plugins/fleet/cypress/e2e/fleet_settings_outputs.cy.ts

@@ -184,8 +184,6 @@ describe('Outputs', () => {
         // Compression
         cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_CODEC_INPUT).should('not.exist');
         cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_SWITCH).click();
-        cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_LEVEL_INPUT).should('not.exist');
-        cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_CODEC_INPUT).select('gzip');
         cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_LEVEL_INPUT).should('exist');
         cy.getBySel(SETTINGS_OUTPUTS_KAFKA.COMPRESSION_LEVEL_INPUT).select('1');
 

x-pack/plugins/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/output_form_kafka_compression.tsx

@@ -21,12 +21,12 @@ export const OutputFormKafkaCompression: React.FunctionComponent<{
 
   const kafkaCompressionTypeOptions = useMemo(
     () =>
-      (Object.keys(kafkaCompressionType) as Array<keyof typeof kafkaCompressionType>).map(
-        (key) => ({
+      (Object.keys(kafkaCompressionType) as Array<keyof typeof kafkaCompressionType>)
+        .filter((c) => c !== 'None')
+        .map((key) => ({
           text: kafkaCompressionType[key],
           label: kafkaCompressionType[key],
-        })
-      ),
+        })),
     []
   );
 

x-pack/plugins/fleet/public/applications/fleet/sections/settings/components/edit_output_flyout/use_output_form.tsx

@@ -391,7 +391,9 @@ export function useOutputForm(onSucess: () => void, output?: Output) {
     isDisabled('compression_level')
   );
   const kafkaCompressionCodecInput = useInput(
-    kafkaOutput?.compression ?? kafkaCompressionType.None,
+    kafkaOutput?.compression && kafkaOutput.compression !== kafkaCompressionType.None
+      ? kafkaOutput.compression
+      : kafkaCompressionType.Gzip,
     undefined,
     isDisabled('compression')
   );
@@ -642,8 +644,11 @@ export function useOutputForm(onSucess: () => void, output?: Output) {
               client_id: kafkaClientIdInput.value || undefined,
               version: kafkaVersionInput.value,
               ...(kafkaKeyInput.value ? { key: kafkaKeyInput.value } : {}),
-              compression: kafkaCompressionCodecInput.value,
-              ...(kafkaCompressionCodecInput.value === kafkaCompressionType.Gzip
+              compression: kafkaCompressionInput.value
+                ? kafkaCompressionCodecInput.value
+                : kafkaCompressionType.None,
+              ...(kafkaCompressionInput.value &&
+              kafkaCompressionCodecInput.value === kafkaCompressionType.Gzip
                 ? {
                     compression_level: parseIntegerIfStringDefined(
                       kafkaCompressionLevelInput.value
@@ -785,6 +790,7 @@ export function useOutputForm(onSucess: () => void, output?: Output) {
     loadBalanceEnabledInput.value,
     typeInput.value,
     kafkaSslCertificateAuthoritiesInput.value,
+    kafkaCompressionInput.value,
     nameInput.value,
     kafkaHostsInput.value,
     defaultOutputInput.value,

x-pack/plugins/fleet/server/services/agent_policies/full_agent_policy.ts

@@ -20,7 +20,7 @@ import type {
 } from '../../types';
 import type { FullAgentPolicyOutputPermissions, PackageInfo } from '../../../common/types';
 import { agentPolicyService } from '../agent_policy';
-import { dataTypes, outputType } from '../../../common/constants';
+import { dataTypes, kafkaCompressionType, outputType } from '../../../common/constants';
 import { DEFAULT_OUTPUT } from '../../constants';
 
 import { getPackageInfo } from '../epm/packages';
@@ -336,7 +336,7 @@ export function transformOutputToFullPolicyOutput(
       version,
       key,
       compression,
-      compression_level,
+      ...(compression === kafkaCompressionType.Gzip ? { compression_level } : {}),
       ...(username ? { username } : {}),
       ...(password ? { password } : {}),
       ...(sasl ? { sasl } : {}),

x-pack/plugins/fleet/server/services/output.ts

@@ -764,6 +764,11 @@ class OutputService {
         ) {
           updateData.compression_level = 4;
         }
+        if (data.compression && data.compression !== kafkaCompressionType.Gzip) {
+          // Clear compression level if compression is not gzip
+          updateData.compression_level = null;
+        }
+
         if (!data.client_id) {
           updateData.client_id = 'Elastic';
         }