Feature: Modify Default Producer Compression Type

backend/pkg/api/handle_producer_compression_type.go

@@ -0,0 +1,33 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file https://github.com/redpanda-data/redpanda/blob/dev/licenses/bsl.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package api
+
+import (
+	"net/http"
+
+	"github.com/cloudhut/common/rest"
+	"go.uber.org/zap"
+)
+
+// handleGetProducerCompressionType returns the default producer compression type
+func (api *API) handleGetProducerCompressionType() http.HandlerFunc {
+	type response struct {
+		ProducerCompressionType string `json:"producerCompressionType"`
+	}
+
+	return func(w http.ResponseWriter, r *http.Request) {
+		producerCompressionType := api.Cfg.Kafka.PComp.Value
+		logger := api.Logger.With(zap.String("producer_compression_type", producerCompressionType))
+
+		rest.SendResponse(w, r, logger, http.StatusOK, &response{
+			ProducerCompressionType: producerCompressionType,
+		})
+	}
+}

backend/pkg/api/routes.go

@@ -477,6 +477,9 @@ func (api *API) routes() *chi.Mux {
 				r.Patch("/operations/reassign-partitions", api.handlePatchPartitionAssignments())
 				r.Patch("/operations/configs", api.handlePatchConfigs())
 
+				// Producer Compression Type Default
+				r.Get("/producer-compression-type", api.handleGetProducerCompressionType())
+
 				// Schema Registry
 				r.Get("/schema-registry/mode", api.handleGetSchemaRegistryMode())
 				r.Get("/schema-registry/config", api.handleGetSchemaRegistryConfig())

backend/pkg/config/kafka.go

@@ -23,6 +23,7 @@ type Kafka struct {
 
 	// Schema Registry
 	Schema      Schema  `yaml:"schemaRegistry"`
+	PComp       PComp   `yaml:"producerCompressionType"`
 	Protobuf    Proto   `yaml:"protobuf"`
 	MessagePack Msgpack `yaml:"messagePack"`
 	Cbor        Cbor    `yaml:"cbor"`

backend/pkg/config/pcomp.go

@@ -0,0 +1,20 @@
+// Copyright 2024 Redpanda Data, Inc.
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.md
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0
+
+package config
+
+// PComp represents the producer compression type default.
+type PComp struct {
+	Value string `yaml:"value"`
+}
+
+// SetDefaults for the producer compression type configuration.
+func (c *PComp) SetDefaults() {
+	c.Value = "snappy"
+}

frontend/src/components/pages/topics/Topic.Produce.tsx

@@ -87,7 +87,7 @@ const PublishTopicForm: FC<{ topicName: string }> = observer(({ topicName }) =>
     } = useForm<Inputs>({
         defaultValues: {
             partition: -1,
-            compressionType: CompressionType.SNAPPY,
+            compressionType: api.producerCompressionType,
             headers: [],
             key: {
                 data: '',
@@ -583,6 +583,7 @@ export class TopicProducePage extends PageComponent<{ topicName: string }> {
 
     refreshData(force?: boolean) {
         api.refreshSchemaSubjects(force);
+        api.refreshProducerCompressionType(force);
     }
 
     render() {

frontend/src/state/backendApi.ts

@@ -79,6 +79,7 @@ import {
     PatchTopicConfigsRequest,
     Payload,
     ProduceRecordsResponse,
+    ProducerCompressionTypeResponse,
     PublishRecordsRequest,
     QuotaResponse,
     ResourceConfig,
@@ -324,6 +325,8 @@ const apiStore = {
     topicConsumers: new Map<string, TopicConsumer[]>(),
     topicAcls: new Map<string, GetAclOverviewResponse | null>(),
 
+    producerCompressionType: undefined as ProtoCompressionType | undefined,
+
     serviceAccounts: undefined as GetUsersResponse | undefined | null,
     ACLs: undefined as GetAclOverviewResponse | undefined | null,
 
@@ -458,6 +461,31 @@ const apiStore = {
             }, addError);
     },
 
+    refreshProducerCompressionType(force?: boolean) {
+        cachedApiRequest<ProducerCompressionTypeResponse>(`${appConfig.restBasePath}/producer-compression-type`, force)
+            .then(r => {
+                if (r.producerCompressionType!=null && r.producerCompressionType.includes('snappy')) {
+                    this.producerCompressionType = ProtoCompressionType.SNAPPY;
+                    return
+                } if (r.producerCompressionType!=null && r.producerCompressionType?.includes('uncompressed')) {
+                    this.producerCompressionType = ProtoCompressionType.UNCOMPRESSED;
+                    return
+                } if (r.producerCompressionType!=null && r.producerCompressionType?.includes('gzip')) {
+                    this.producerCompressionType = ProtoCompressionType.GZIP;
+                    return
+                } if (r.producerCompressionType!=null && r.producerCompressionType?.includes('lz4')) {
+                    this.producerCompressionType = ProtoCompressionType.LZ4;
+                    return
+                } if (r.producerCompressionType!=null && r.producerCompressionType?.includes('zstd')) {
+                    this.producerCompressionType = ProtoCompressionType.ZSTD;
+                    return
+                } else {
+                    this.producerCompressionType = ProtoCompressionType.SNAPPY //default to snappy
+                    return
+                }
+            }, addError);
+    },
+
     refreshTopicPermissions(topicName: string, force?: boolean) {
         if (!AppFeatures.SINGLE_SIGN_ON) return; // without SSO there can't be a permissions endpoint
         if (this.userData?.user?.providerID == -1) return; // debug user

frontend/src/state/restInterfaces.ts

@@ -268,7 +268,10 @@ export interface TopicDocumentationResponse {
     documentation: TopicDocumentation;
 }
 
-
+// GET /producer-compression-type
+export type ProducerCompressionTypeResponse = {
+    producerCompressionType: string | null;
+};
 
 
 export interface GroupMemberAssignment {