Merge pull request #358 from Consdata/IKC-387-resend-event

pbelke · web-flow · commit d66d55be6a87 · 2024-06-10T09:31:49.000+02:00
IKC-387 Resend event
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/serde/SchemaMessageSerde.java b/kouncil-backend/src/main/java/com/consdata/kouncil/serde/SchemaMessageSerde.java
@@ -11,6 +11,7 @@
 
 @Service
 public class SchemaMessageSerde {
+
     public DeserializedData deserialize(SchemaAwareCluster schemaAwareCluster, Bytes payload, KouncilSchemaMetadata kouncilSchemaMetadata) {
         MessageFormat messageFormat = schemaAwareCluster.getSchemaRegistryFacade().getSchemaFormat(kouncilSchemaMetadata);
         MessageFormatter formatter = schemaAwareCluster.getFormatter(messageFormat);
@@ -19,6 +20,12 @@ public DeserializedData deserialize(SchemaAwareCluster schemaAwareCluster, Bytes
                 .deserialized(formatter.deserialize(DeserializationData.builder()
                         .value(payload.get())
                         .topicName(kouncilSchemaMetadata.getSchemaTopic())
+                        .useLogicalTypesConversions(true)
+                        .build()))
+                .originalValue(formatter.deserialize(DeserializationData.builder()
+                        .value(payload.get())
+                        .topicName(kouncilSchemaMetadata.getSchemaTopic())
+                        .useLogicalTypesConversions(false)
                         .build()))
                 .messageFormat(formatter.getFormat())
                 .schemaId(kouncilSchemaMetadata.getSchemaId())
@@ -35,7 +42,7 @@ public Bytes serialize(SchemaAwareCluster schemaAwareCluster, String payload, Ko
                         .topicName(kouncilSchemaMetadata.getSchemaTopic())
                         .schema(schema)
                         .isKey(kouncilSchemaMetadata.isKey())
-                .build()
+                        .build()
         );
     }
 }
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/serde/deserialization/DeserializationData.java b/kouncil-backend/src/main/java/com/consdata/kouncil/serde/deserialization/DeserializationData.java
@@ -8,4 +8,5 @@
 public class DeserializationData {
     String topicName;
     byte[] value;
+    boolean useLogicalTypesConversions;
 }
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/serde/deserialization/DeserializedData.java b/kouncil-backend/src/main/java/com/consdata/kouncil/serde/deserialization/DeserializedData.java
@@ -8,6 +8,7 @@
 @Builder
 public class DeserializedData {
     String deserialized;
+    String originalValue;
     MessageFormat messageFormat;
     Integer schemaId;
 }
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/serde/formatter/schema/AvroMessageFormatter.java b/kouncil-backend/src/main/java/com/consdata/kouncil/serde/formatter/schema/AvroMessageFormatter.java
@@ -19,19 +19,19 @@
 @Slf4j
 public class AvroMessageFormatter implements MessageFormatter {
 
-    private final KafkaAvroDeserializer avroDeserializer;
     private final KafkaAvroSerializer avroSerializer;
+    private final SchemaRegistryClient client;
 
     public AvroMessageFormatter(SchemaRegistryClient client) {
-        this.avroDeserializer = new KafkaAvroDeserializer(client);
         this.avroSerializer = new KafkaAvroSerializer(client);
-        this.configureDeserializer();
+        this.client = client;
     }
 
     @Override
     public String deserialize(DeserializationData deserializationData) {
-        Object deserialized = avroDeserializer.deserialize(deserializationData.getTopicName(), deserializationData.getValue());
-        return deserialized.toString();
+        KafkaAvroDeserializer avroDeserializer = new KafkaAvroDeserializer(client);
+        this.configureDeserializer(avroDeserializer, deserializationData.isUseLogicalTypesConversions());
+        return avroDeserializer.deserialize(deserializationData.getTopicName(), deserializationData.getValue()).toString();
     }
 
     @Override
@@ -64,12 +64,12 @@ private void configureSerializer(SerializationData serializationData) {
         );
     }
 
-    private void configureDeserializer() {
+    private void configureDeserializer(KafkaAvroDeserializer avroDeserializer, boolean useLogicalTypes) {
         avroDeserializer.configure(
                 Map.of(
                         AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "needed_in_runtime_but_not_used",
                         AbstractKafkaSchemaSerDeConfig.USE_LATEST_VERSION, true,
-                        KafkaAvroDeserializerConfig.AVRO_USE_LOGICAL_TYPE_CONVERTERS_CONFIG, true
+                        KafkaAvroDeserializerConfig.AVRO_USE_LOGICAL_TYPE_CONVERTERS_CONFIG, useLogicalTypes
                 ),
                 false
         );
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicMessage.java b/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicMessage.java
@@ -1,11 +1,10 @@
 package com.consdata.kouncil.topic;
 
 import com.consdata.kouncil.serde.MessageFormat;
+import java.util.List;
 import lombok.Builder;
 import lombok.Data;
 
-import java.util.List;
-
 @Data
 @Builder
 public class TopicMessage {
@@ -14,6 +13,7 @@ public class TopicMessage {
     private String key;
     private MessageFormat keyFormat;
     private String value;
+    private String originalValue;
     private MessageFormat valueFormat;
     private long timestamp;
     private int partition;
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicService.java b/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicService.java
@@ -196,6 +196,7 @@ private void pollMessages(String clusterId, int limit, KafkaConsumer<Bytes, Byte
                             .key(deserializedMessage.getKeyData().getDeserialized())
                             .keyFormat(deserializedMessage.getKeyData().getMessageFormat())
                             .value(deserializedMessage.getValueData().getDeserialized())
+                            .originalValue(deserializedMessage.getValueData().getOriginalValue())
                             .valueFormat(deserializedMessage.getValueData().getMessageFormat())
                             .offset(consumerRecord.offset())
                             .partition(consumerRecord.partition())
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/json-grid-data.ts b/kouncil-frontend/apps/kouncil/src/app/topic/json-grid-data.ts
@@ -3,6 +3,7 @@ import {MessageFormat} from '@app/schema-registry';
 
 export interface JsonGridData {
   value: string;
+  originalValue: string;
   valueFormat: MessageFormat;
   valueJson: Record<string, unknown>;
   partition: number | null;
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/json-grid.spec.ts b/kouncil-frontend/apps/kouncil/src/app/topic/json-grid.spec.ts
@@ -73,6 +73,7 @@ describe('JsonGrid', () => {
       'kouncilTimestamp',
       'kouncilTimestampEpoch',
       'kouncilValue',
+      'kouncilOriginalValue',
       'kouncilValueFormat',
       'kouncilValueJson',
       'headers']);
@@ -124,6 +125,7 @@ describe('JsonGrid', () => {
       'kouncilTimestamp',
       'kouncilTimestampEpoch',
       'kouncilValue',
+      'kouncilOriginalValue',
       'kouncilValueFormat',
       'kouncilValueJson',
       'headers']);
@@ -174,6 +176,7 @@ describe('JsonGrid', () => {
       'kouncilTimestamp',
       'kouncilTimestampEpoch',
       'kouncilValue',
+      'kouncilOriginalValue',
       'kouncilValueFormat',
       'kouncilValueJson',
       'headers']);
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/json-grid.ts b/kouncil-frontend/apps/kouncil/src/app/topic/json-grid.ts
@@ -102,6 +102,7 @@ export class JsonGrid {
         row['kouncilTimestamp'] = this.formatTimestamp(object.timestamp);
         row['kouncilTimestampEpoch'] = object.timestamp;
         row['kouncilValue'] = object.value;
+        row['kouncilOriginalValue'] = object.originalValue;
         row['kouncilValueFormat'] = object.valueFormat;
         row['kouncilValueJson'] = object.valueJson;
         row['headers'] = object.headers;
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/topic.component.ts b/kouncil-frontend/apps/kouncil/src/app/topic/topic.component.ts
@@ -177,6 +177,7 @@ export class TopicComponent extends AbstractTableComponent implements OnInit, On
     const messageData = {
       value: event.kouncilValueJson && Object.keys(event.kouncilValueJson).length > 0 ?
         event.kouncilValueJson : event.kouncilValue,
+      originalValue: event.kouncilOriginalValue,
       valueFormat: event.kouncilValueFormat,
       headers: event.headers,
       key: event.kouncilKeyJson && Object.keys(event.kouncilKeyJson).length > 0 ?
@@ -212,6 +213,7 @@ export class TopicComponent extends AbstractTableComponent implements OnInit, On
     topicMessages.messages.forEach((message: MessageData) =>
       values.push({
         value: message.value,
+        originalValue: message.originalValue,
         valueFormat: message.valueFormat,
         valueJson: TopicComponent.tryParseJson(message.value),
         partition: message.partition,
diff --git a/kouncil-frontend/libs/feat-send/src/lib/send/send.component.ts b/kouncil-frontend/libs/feat-send/src/lib/send/send.component.ts
@@ -45,7 +45,7 @@ declare let monaco: any;
 
         <div class="drawer-section-title">
           Headers
-          <button type="button" class="small-button" mat-button disableRipple
+          <button type="button" class="small-button" mat-button [disableRipple]="true"
                   (click)="addHeader(messageData.headers)">
             +
           </button>
@@ -59,7 +59,7 @@ declare let monaco: any;
             <input class="header" [(ngModel)]="header.value" placeholder="Header value" matInput
                    type="text" name="header-value-{{ i }}"/>
           </mat-form-field>
-          <button type="button" class="small-button" mat-button disableRipple
+          <button type="button" class="small-button" mat-button [disableRipple]="true"
                   (click)="removeHeader(i, messageData.headers)">
             -
           </button>
@@ -79,11 +79,11 @@ declare let monaco: any;
         <mat-form-field [appearance]="'outline'" class="count">
           <input matInput type="number" min="1" [formControl]="countControl" name="count"/>
           <div matSuffix>
-            <button type="button" class="small-button" mat-button disableRipple
+            <button type="button" class="small-button" mat-button [disableRipple]="true"
                     (click)="decreaseCount()">
               -
             </button>
-            <button type="button" class="small-button" mat-button disableRipple
+            <button type="button" class="small-button" mat-button [disableRipple]="true"
                     (click)="increaseCount()">
               +
             </button>
@@ -93,11 +93,11 @@ declare let monaco: any;
         <span class="spacer"></span>
 
         <div class="actions">
-          <button type="button" mat-dialog-close mat-button disableRipple
+          <button type="button" mat-dialog-close mat-button [disableRipple]="true"
                   class="action-button-white">
             Cancel
           </button>
-          <button mat-button disableRipple
+          <button mat-button [disableRipple]="true"
                   class="action-button-black"
                   type="submit"
                   [disabled]="isSendButtonDisabled">
@@ -136,13 +136,16 @@ export class SendComponent implements OnDestroy {
             map(exampleData => ({
                 ...messageData,
                 key: messageData.key ?? JSON.stringify(exampleData.exampleKey),
-                value: messageData.value ? JSON.stringify(messageData.value, null, 2) :
-                  JSON.stringify(exampleData.exampleValue, null, 2)
+                value: messageData.originalValue
+                  ? messageData.originalValue
+                  : (messageData.value
+                    ? messageData.value
+                    : JSON.stringify(exampleData.exampleValue, null, 2))
               })
             )),
           of({
               ...messageData,
-              value: messageData.value ? JSON.stringify(messageData.value, null, 2) : messageData.value
+              value: messageData.originalValue ? JSON.stringify(messageData.originalValue, null, 2) : messageData.value
             }
           ));
       }
@@ -252,11 +255,15 @@ export class SendComponent implements OnDestroy {
     this.countControl.reset(1);
   }
 
-  addHeader(headers: MessageDataHeader[]): void {
-    headers.push({key: '', value: ''} as MessageDataHeader);
+  addHeader(headers: MessageDataHeader[] | undefined): void {
+    if (headers) {
+      headers.push({key: '', value: ''} as MessageDataHeader);
+    }
   }
 
-  removeHeader(i: number, headers: MessageDataHeader[]): void {
-    headers.splice(i, 1);
+  removeHeader(i: number, headers: MessageDataHeader[] | undefined): void {
+    if (headers) {
+      headers.splice(i, 1);
+    }
   }
 }
diff --git a/kouncil-frontend/libs/message-data/src/lib/message-data/message-data.ts b/kouncil-frontend/libs/message-data/src/lib/message-data/message-data.ts
@@ -5,6 +5,7 @@ export interface MessageData {
   key: string;
   keyFormat: MessageFormat;
   value: string;
+  originalValue: string;
   valueFormat: MessageFormat;
   offset: number | null;
   partition: number | null;

Original file line number	Diff line number	Diff line change
`@@ -8,4 +8,5 @@`
`8`	`8`	`public class DeserializationData {`
`9`	`9`	`String topicName;`
`10`	`10`	`byte[] value;`
	`11`	`+ boolean useLogicalTypesConversions;`
`11`	`12`	`}`
Original file line number	Diff line number	Diff line change
`@@ -8,6 +8,7 @@`
`8`	`8`	`@Builder`
`9`	`9`	`public class DeserializedData {`
`10`	`10`	`String deserialized;`
	`11`	`+ String originalValue;`
`11`	`12`	`MessageFormat messageFormat;`
`12`	`13`	`Integer schemaId;`
`13`	`14`	`}`