Merge pull request #470 from Consdata/IKC-470

pbelke · web-flow · commit d8f3e38447c3 · 2025-10-18T14:44:22.000+02:00
IKC-470 reset form when cluster was changed
diff --git a/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicController.java b/kouncil-backend/src/main/java/com/consdata/kouncil/topic/TopicController.java
@@ -1,5 +1,7 @@
 package com.consdata.kouncil.topic;
 
+import static com.consdata.kouncil.model.admin.SystemFunctionNameConstants.TOPIC_SEND_MESSAGE;
+
 import com.consdata.kouncil.KouncilRuntimeException;
 import com.consdata.kouncil.logging.EntryExitLogger;
 import com.consdata.kouncil.model.admin.SystemFunctionNameConstants;
@@ -39,7 +41,7 @@ public TopicMessagesDto getTopicMessages(@PathVariable("topicName") String topic
         return topicService.getTopicMessages(topicName, partitions, pageParam, limitParam, beginningTimestampMillis, endTimestampMillis, offset, serverId);
     }
 
-    @RolesAllowed(SystemFunctionNameConstants.TOPIC_SEND_MESSAGE)
+    @RolesAllowed(TOPIC_SEND_MESSAGE)
     @PostMapping("/send/{topicName}/{count}")
     @EntryExitLogger
     public void send(@PathVariable("topicName") String topicName,
@@ -87,4 +89,11 @@ public TopicData getTopicData(@PathVariable("topicName") String topicName, @Requ
     public void removeTopic(@PathVariable("topicName") String topicName, @RequestParam("serverId") String serverId) throws KouncilRuntimeException {
         topicService.removeTopic(topicName, serverId);
     }
+
+    @RolesAllowed(SystemFunctionNameConstants.TOPIC_MESSAGES)
+    @GetMapping("/is-topic-exist/{topicName}")
+    @EntryExitLogger
+    public void isTopicExist(@PathVariable("topicName") String topicName, @RequestParam("serverId") String serverId) {
+        topicService.isTopicExist(topicName, serverId);
+    }
 }
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
@@ -50,8 +50,6 @@
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestParam;
 
 @Slf4j
 @Service
@@ -69,14 +67,14 @@ public class TopicService {
     @Value("${resendHeadersToKeep:}")
     private String[] resendHeadersToKeep;
 
-    TopicMessagesDto getTopicMessages(@PathVariable("topicName") String topicName,
-            @PathVariable("partition") String partitions,
-            @RequestParam("page") String pageParam,
-            @RequestParam("limit") String limitParam,
-            @RequestParam(value = "beginningTimestampMillis", required = false) Long beginningTimestampMillis,
-            @RequestParam(value = "endTimestampMillis", required = false) Long endTimestampMillis,
-            @RequestParam(value = "offset", required = false) Long offset,
-            @RequestParam("serverId") String serverId) {
+    TopicMessagesDto getTopicMessages(String topicName,
+                                      String partitions,
+                                      String pageParam,
+                                      String limitParam,
+                                      Long beginningTimestampMillis,
+                                      Long endTimestampMillis,
+                                      Long offset,
+                                      String serverId) {
         messagesHelper.validateTopics(serverId, singletonList(topicName));
         int limit = Integer.parseInt(limitParam); // per partition!
         long page = Long.parseLong(pageParam); // per partition!
@@ -368,4 +366,8 @@ public void removeTopic(String topicName, String serverId) {
             throw new KouncilRuntimeException(e.getMessage());
         }
     }
+
+    public void isTopicExist(String topicName, String serverId) {
+        messagesHelper.validateTopics(serverId, List.of(topicName));
+    }
 }
diff --git a/kouncil-frontend/apps/kouncil/src/app/schemas/form/form/schema-form.component.ts b/kouncil-frontend/apps/kouncil/src/app/schemas/form/form/schema-form.component.ts
@@ -1,20 +1,29 @@
-import { ChangeDetectorRef, Component, EventEmitter, Input, OnInit, Output } from '@angular/core';
+import {
+  ChangeDetectorRef,
+  Component,
+  EventEmitter,
+  Input,
+  OnDestroy,
+  OnInit,
+  Output
+} from '@angular/core';
 import { ProgressBarService, ViewMode } from '@app/common-utils';
 import {
   Compatibility,
   MessageFormat,
   Schema,
-  SchemaRegistryService,
+  SchemaRegistryService, SchemaStateService,
   SubjectType
 } from '@app/schema-registry';
 import { ServersService } from '@app/common-servers';
-import { ActivatedRoute } from '@angular/router';
+import {ActivatedRoute, Router} from '@angular/router';
 import { TopicsService } from '@app/feat-topics';
 import { SelectableItem } from '@app/common-components';
 import { Topics } from '@app/common-model';
 import { first } from 'rxjs/operators';
 import { MatSelectChange } from '@angular/material/select';
 import { FormControl, FormGroup, Validators } from '@angular/forms';
+import {Subscription} from 'rxjs';
 
 @Component({
   selector: 'app-schema-form',
@@ -83,7 +92,7 @@ import { FormControl, FormGroup, Validators } from '@angular/forms';
   `,
   styleUrls: ['./schema-form.component.scss']
 })
-export class SchemaFormComponent implements OnInit {
+export class SchemaFormComponent implements OnInit, OnDestroy {
 
   model: Schema;
   subjectTypes: SelectableItem[] = Object.keys(SubjectType).map(format => new SelectableItem(format, format, false));
@@ -109,13 +118,16 @@ export class SchemaFormComponent implements OnInit {
     compatibility: new FormControl(),
     plainTextSchema: new FormControl('')
   });
+  private readonly subscription: Subscription = new Subscription();
 
   constructor(private schemaRegistry: SchemaRegistryService,
               private servers: ServersService,
               private progressBarService: ProgressBarService,
               private route: ActivatedRoute,
               private topicsService: TopicsService,
-              private cdr: ChangeDetectorRef) {
+              private cdr: ChangeDetectorRef,
+              private schemaStateService: SchemaStateService,
+              private router: Router) {
   }
 
   ngOnInit(): void {
@@ -142,6 +154,16 @@ export class SchemaFormComponent implements OnInit {
       this.topics = topics.topics.map((tm) => new SelectableItem(tm.name, tm.name, false));
       this.progressBarService.setProgress(false);
     });
+
+    this.subscription.add(this.schemaStateService.isSchemaConfigured$(this.servers.getSelectedServerId()).subscribe((hasSchemaConnected) => {
+        if (!hasSchemaConnected) {
+          this.router.navigate(['/schemas']);
+        }
+      }));
+  }
+
+  ngOnDestroy(): void  {
+    this.subscription.unsubscribe();
   }
 
   private defineDisabled() {
diff --git a/kouncil-frontend/apps/kouncil/src/app/schemas/list/schemas.component.ts b/kouncil-frontend/apps/kouncil/src/app/schemas/list/schemas.component.ts
@@ -1,5 +1,5 @@
 import {Component, OnInit} from '@angular/core';
-import {Schema, SchemaRegistryService} from '@app/schema-registry';
+import {Schema, SchemaRegistryService, SchemaStateService} from '@app/schema-registry';
 import {ServersService} from '@app/common-servers';
 import {AbstractTableComponent, SelectableItem, TableColumn} from '@app/common-components';
 import {ProgressBarService, SnackBarComponent, SnackBarData, SnackBarType} from '@app/common-utils';
@@ -39,7 +39,7 @@ const log = LoggerFactory.getLogger('SchemasComponent');
             Clear filters
           </button>
 
-          <button mat-button *ngIf="authService.canAccess([SystemFunctionName.SCHEMA_CREATE])"
+          <button mat-button *ngIf="authService.canAccess([SystemFunctionName.SCHEMA_CREATE]) && serverHasSchemaConnected"
                   class="action-button-blue" [routerLink]="['/schemas/create']">
             Add new schema
           </button>
@@ -151,6 +151,7 @@ export class SchemasComponent extends AbstractTableComponent implements OnInit {
   topicFilterForm: FormGroup = new FormGroup({
     topicFilterControl: new FormControl()
   });
+  serverHasSchemaConnected: boolean = false;
 
   constructor(private progressBarService: ProgressBarService,
               private schemaRegistry: SchemaRegistryService,
@@ -159,7 +160,8 @@ export class SchemasComponent extends AbstractTableComponent implements OnInit {
               private snackbar: MatSnackBar,
               protected authService: AuthService,
               private topicsService: TopicsService,
-              private router: Router) {
+              private router: Router,
+              private schemaStateService: SchemaStateService) {
     super();
   }
 
@@ -179,6 +181,10 @@ export class SchemasComponent extends AbstractTableComponent implements OnInit {
       this.filtered = data;
       this.progressBarService.setProgress(false);
     });
+
+    this.schemaStateService.isSchemaConfigured$(this.servers.getSelectedServerId()).subscribe(hasSchemaConnected=>{
+      this.serverHasSchemaConnected = hasSchemaConnected;
+    });
   }
 
   deleteSchema(subject: string, version: string): void {
diff --git a/kouncil-frontend/apps/kouncil/src/app/toolbar/toolbar.component.ts b/kouncil-frontend/apps/kouncil/src/app/toolbar/toolbar.component.ts
@@ -113,6 +113,7 @@ export class ToolbarComponent implements OnInit, AfterViewInit, OnDestroy {
 
   public serverSelectionChanged(): void {
     localStorage.setItem('lastSelectedServer', this.servers.getSelectedServerId());
+    this.servers.selectedServerChanged();
     this.router.navigate([this.router.url]);
   }
 
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/topic.backend.service.ts b/kouncil-frontend/apps/kouncil/src/app/topic/topic.backend.service.ts
@@ -5,6 +5,7 @@ import {TopicMessages} from './topic-messages';
 import {Page} from './page';
 import {HttpClient, HttpParams} from '@angular/common/http';
 import {ProgressBarService} from '@app/common-utils';
+import {map} from 'rxjs/operators';
 
 @Injectable({
   providedIn: 'root'
@@ -15,16 +16,16 @@ export class TopicBackendService implements TopicService {
   partitionEndOffsets: { [key: number]: number } = {};
   partitions?: number[];
   selectedPartition?: string;
-  private convertTopicMessagesJsonToGrid$: Subject<TopicMessages> = new Subject<TopicMessages>();
-  private numberOfPartitionsChanged$: Subject<number> = new Subject<number>();
+  private readonly convertTopicMessagesJsonToGrid$: Subject<TopicMessages> = new Subject<TopicMessages>();
+  private readonly numberOfPartitionsChanged$: Subject<number> = new Subject<number>();
   // eslint-disable-next-line rxjs/no-exposed-subjects
   protected paginationChanged$: BehaviorSubject<Page> = new BehaviorSubject<Page>({pageNumber: 1, size: 10});
 
   constructor(protected http: HttpClient, protected progressBarService: ProgressBarService) {
   }
 
   getMessages(serverId: string, topicName: string, offset?: number): void {
-    let url;
+    let url: string;
     if (typeof this.selectedPartition !== 'undefined') {
       url = `/api/topic/messages/${topicName}/${this.selectedPartition}`;
     } else {
@@ -92,4 +93,12 @@ export class TopicBackendService implements TopicService {
   goToOffset(serverId: string, topicName: string, offset: number): void {
     this.getMessages(serverId, topicName, offset);
   }
+
+  isTopicExist$(serverId: string, topicName: string): Observable<boolean>{
+    const params = new HttpParams().set('serverId', serverId);
+
+    return this.http.get<void>(`/api/topic/is-topic-exist/${topicName}`, {params}).pipe(map(() => {
+      return true;
+    }));
+  }
 }
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/topic.component.ts b/kouncil-frontend/apps/kouncil/src/app/topic/topic.component.ts
@@ -13,7 +13,11 @@ import {Observable, Subscription} from 'rxjs';
 import {JsonGrid} from './json-grid';
 import {TopicMessages} from './topic-messages';
 import {MessageData, MessageDataService} from '@app/message-data';
-import {DrawerService, ProgressBarService, SearchService} from '@app/common-utils';
+import {
+  DrawerService,
+  ProgressBarService,
+  SearchService,
+} from '@app/common-utils';
 import {ServersService} from '@app/common-servers';
 import {SendComponent} from '@app/feat-send';
 import {AbstractTableComponent, TableColumn} from '@app/common-components';
@@ -83,17 +87,17 @@ export class TopicComponent extends AbstractTableComponent implements OnInit, On
   paging$: Observable<Page> = this.topicService.getPagination$();
 
   constructor(
-    private route: ActivatedRoute,
-    private searchService: SearchService,
-    private jsonGrid: JsonGrid,
-    private titleService: Title,
-    private progressBarService: ProgressBarService,
-    private topicService: TopicService,
-    private drawerService: DrawerService,
-    private servers: ServersService,
-    private messageDataService: MessageDataService,
-    private router: Router,
-    private location: Location
+    private readonly route: ActivatedRoute,
+    private readonly searchService: SearchService,
+    private readonly jsonGrid: JsonGrid,
+    private readonly titleService: Title,
+    private readonly progressBarService: ProgressBarService,
+    private readonly topicService: TopicService,
+    private readonly drawerService: DrawerService,
+    private readonly servers: ServersService,
+    private readonly messageDataService: MessageDataService,
+    private readonly router: Router,
+    private readonly location: Location
   ) {
     super();
     this.jsonToGridSubscription = this.topicService
@@ -116,10 +120,7 @@ export class TopicComponent extends AbstractTableComponent implements OnInit, On
     this.progressBarService.setProgress(true);
     this.route.params.subscribe((params) => {
       this.topicName = params['topic'];
-      this.topicService.getMessages(
-        this.servers.getSelectedServerId(),
-        this.topicName
-      );
+      this.getMessageIfTopicExist();
       this.titleService.setTitle(this.topicName + ' Kouncil');
       this.paused = true;
     });
@@ -336,4 +337,11 @@ export class TopicComponent extends AbstractTableComponent implements OnInit, On
     const urlTree = this.router.createUrlTree([], {relativeTo: this.route, queryParams: {page}});
     this.location.go(urlTree.toString());
   }
+
+  private getMessageIfTopicExist() {
+    this.topicService.isTopicExist$(this.servers.getSelectedServerId(), this.topicName).subscribe({
+      next: ()=> this.topicService.getMessages(this.servers.getSelectedServerId(), this.topicName),
+      error: ()=> this.router.navigate(['/topics'])
+    });
+  }
 }
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/topic.demo.service.ts b/kouncil-frontend/apps/kouncil/src/app/topic/topic.demo.service.ts
@@ -4,6 +4,7 @@ import {TopicBackendService} from './topic.backend.service';
 import {MessageData, MessageDataHeader} from '@app/message-data';
 import {Crypto, RandomUtils} from '@app/common-utils';
 import {demoTopics} from '@app/feat-topics';
+import {Observable, of} from 'rxjs';
 
 @Injectable()
 export class TopicDemoService extends TopicBackendService {
@@ -76,4 +77,8 @@ export class TopicDemoService extends TopicBackendService {
       topicName: ''
     } as MessageData;
   }
+
+  override isTopicExist$(_serverId: string, _topicName: string): Observable<boolean> {
+    return of(true);
+  }
 }
diff --git a/kouncil-frontend/apps/kouncil/src/app/topic/topic.service.ts b/kouncil-frontend/apps/kouncil/src/app/topic/topic.service.ts
@@ -27,6 +27,8 @@ export abstract class TopicService {
   abstract goToOffset(serverId: string, topicName: string, offset?: number): void;
 
   abstract getPagination$(): Observable<Page>;
+
+  abstract isTopicExist$(serverId: string, topicName: string): Observable<boolean>;
 }
 
 export function topicServiceFactory(http: HttpClient, progressBarService: ProgressBarService): TopicService {
diff --git a/kouncil-frontend/apps/kouncil/src/app/track/track-filter/track-filter.component.ts b/kouncil-frontend/apps/kouncil/src/app/track/track-filter/track-filter.component.ts
@@ -1,11 +1,17 @@
-import {Component, OnInit, ViewChild} from '@angular/core';
+import {
+  Component,
+  OnDestroy,
+  OnInit,
+  ViewChild
+} from '@angular/core';
 import {TrackService} from '../track.service';
 import {FormControl, FormGroup, NgForm} from '@angular/forms';
 import {TrackFilter, TrackOperator} from './track-filter';
 import {TopicsService} from '@app/feat-topics';
 import {Topics} from '@app/common-model';
 import {ServersService} from '@app/common-servers';
 import {SelectableItem} from '@app/common-components';
+import {Subscription} from 'rxjs';
 
 @Component({
   selector: 'app-track-filter',
@@ -109,7 +115,7 @@ import {SelectableItem} from '@app/common-components';
   `,
   styleUrls: ['./track-filter.component.scss'],
 })
-export class TrackFilterComponent implements OnInit {
+export class TrackFilterComponent implements OnInit, OnDestroy {
   @ViewChild('filtersForm', {static: false}) filtersForm?: NgForm;
 
   operators: { name: string; index: number }[] = Object.keys(TrackOperator)
@@ -136,11 +142,11 @@ export class TrackFilterComponent implements OnInit {
   correlationTooltip: string = 'Filter messages by specifying the name and value of a message header.\nYou can select the matching method:\n' +
     '~ - contains value\n!~ - does not contain value\nis - exact match\nis not - does not match\nregex - regular expression match ';
 
-  constructor(
-    private trackService: TrackService,
-    private topicsService: TopicsService,
-    private servers: ServersService
-  ) {
+  private readonly subscription: Subscription = new Subscription();
+
+  constructor(private readonly trackService: TrackService,
+              private readonly topicsService: TopicsService,
+              private readonly servers: ServersService) {
     this.trackFilter = this.trackService.getStoredTrackFilter();
   }
 
@@ -153,9 +159,17 @@ export class TrackFilterComponent implements OnInit {
     });
     this.trackFilter = this.trackService.getStoredTrackFilter();
 
-    this.trackService.trackFinished.subscribe(() => {
+    this.subscription.add(this.trackService.trackFinished.subscribe(() => {
       this.loading = false;
-    });
+    }));
+
+    this.subscription.add(this.servers.selectedServerChanged$.subscribe(() => {
+      this.trackService.resetStoredTrackFilter();
+    }));
+  }
+
+  ngOnDestroy(): void {
+    this.subscription.unsubscribe();
   }
 
   toggleAsyncMode(): void {
diff --git a/kouncil-frontend/apps/kouncil/src/app/track/track.service.ts b/kouncil-frontend/apps/kouncil/src/app/track/track.service.ts
diff --git a/kouncil-frontend/libs/common-servers/src/lib/servers/servers.service.ts b/kouncil-frontend/libs/common-servers/src/lib/servers/servers.service.ts
diff --git a/kouncil-frontend/yarn.lock b/kouncil-frontend/yarn.lock

Original file line number	Diff line number	Diff line change
`@@ -113,6 +113,7 @@ export class ToolbarComponent implements OnInit, AfterViewInit, OnDestroy {`
`113`	`113`
`114`	`114`	`public serverSelectionChanged(): void {`
`115`	`115`	`localStorage.setItem('lastSelectedServer', this.servers.getSelectedServerId());`
	`116`	`+ this.servers.selectedServerChanged();`
`116`	`117`	`this.router.navigate([this.router.url]);`
`117`	`118`	`}`
`118`	`119`
Original file line number	Diff line number	Diff line change
`@@ -27,6 +27,8 @@ export abstract class TopicService {`
`27`	`27`	`abstract goToOffset(serverId: string, topicName: string, offset?: number): void;`
`28`	`28`
`29`	`29`	`abstract getPagination$(): Observable<Page>;`
	`30`	`+`
	`31`	`+ abstract isTopicExist$(serverId: string, topicName: string): Observable<boolean>;`
`30`	`32`	`}`
`31`	`33`
`32`	`34`	`export function topicServiceFactory(http: HttpClient, progressBarService: ProgressBarService): TopicService {`