feat: STREAMP-12186: adding disable flags for entity status support

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ConsumerBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2021 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -54,28 +54,32 @@ public Optional<ConsumerBinding> create(ConsumerBinding consumerBinding) throws
     }
     consumerBindingValidator.validateForCreate(consumerBinding);
     consumerBinding.setSpecification(handlerService.handleInsert(consumerBinding));
-    return save(consumerBinding);
+    return saveSpecification(consumerBinding);
   }
 
   @PreAuthorize("hasPermission(#consumerBinding, 'UPDATE')")
   public Optional<ConsumerBinding> update(ConsumerBinding consumerBinding) throws ValidationException {
     val existing = consumerBindingView.get(consumerBinding.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + consumerBinding.getKey() + " because it doesn't exist");
     }
     consumerBindingValidator.validateForUpdate(consumerBinding, existing.get());
     consumerBinding.setSpecification(handlerService.handleUpdate(consumerBinding, existing.get()));
-    return save(consumerBinding);
+    return saveSpecification(consumerBinding);
   }
 
   @PreAuthorize("hasPermission(#consumerBinding, 'UPDATE_STATUS')")
   public Optional<ConsumerBinding> updateStatus(ConsumerBinding consumerBinding, Status status) {
     consumerBinding.setStatus(status);
-    return save(consumerBinding);
+    return saveStatus(consumerBinding);
   }
 
-  private Optional<ConsumerBinding> save(ConsumerBinding consumerBinding) {
-    return Optional.ofNullable(consumerBindingRepository.save(consumerBinding));
+  private Optional<ConsumerBinding> saveSpecification(ConsumerBinding consumerBinding) {
+    return Optional.ofNullable(consumerBindingRepository.saveSpecification(consumerBinding));
+  }
+
+  private Optional<ConsumerBinding> saveStatus(ConsumerBinding consumerBinding) {
+    return Optional.ofNullable(consumerBindingRepository.saveStatus(consumerBinding));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ConsumerService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2022 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,28 +56,32 @@ public Optional<Consumer> create(Consumer consumer) throws ValidationException {
     }
     consumerValidator.validateForCreate(consumer);
     consumer.setSpecification(handlerService.handleInsert(consumer));
-    return save(consumer);
+    return saveSpecification(consumer);
   }
 
   @PreAuthorize("hasPermission(#consumer, 'UPDATE')")
   public Optional<Consumer> update(Consumer consumer) throws ValidationException {
     val existing = consumerView.get(consumer.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + consumer.getKey().getName() + " because it doesn't exist");
     }
     consumerValidator.validateForUpdate(consumer, existing.get());
     consumer.setSpecification(handlerService.handleUpdate(consumer, existing.get()));
-    return save(consumer);
+    return saveSpecification(consumer);
   }
 
   @PreAuthorize("hasPermission(#consumer, 'UPDATE_STATUS')")
   public Optional<Consumer> updateStatus(Consumer consumer, Status status) {
     consumer.setStatus(status);
-    return save(consumer);
+    return saveStatus(consumer);
   }
 
-  private Optional<Consumer> save(Consumer consumer) {
-    return Optional.ofNullable(consumerRepository.save(consumer));
+  private Optional<Consumer> saveSpecification(Consumer consumer) {
+    return Optional.ofNullable(consumerRepository.saveSpecification(consumer));
+  }
+
+  private Optional<Consumer> saveStatus(Consumer consumer) {
+    return Optional.ofNullable(consumerRepository.saveStatus(consumer));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/DomainService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2022 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -68,30 +68,35 @@ public Optional<Domain> create(Domain domain) throws ValidationException {
     }
     domainValidator.validateForCreate(domain);
     domain.setSpecification(handlerService.handleInsert(domain));
-    return save(domain);
+    return saveSpecification(domain);
   }
 
   @PreAuthorize("hasPermission(#domain, 'UPDATE')")
   public Optional<Domain> update(Domain domain) throws ValidationException {
     val existing = domainView.get(domain.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + domain.getKey().getName() + " because it doesn't exist");
     }
     domainValidator.validateForUpdate(domain, existing.get());
     domain.setSpecification(handlerService.handleUpdate(domain, existing.get()));
-    return save(domain);
+    return saveSpecification(domain);
   }
 
   @PreAuthorize("hasPermission(#domain, 'UPDATE_STATUS')")
   public Optional<Domain> updateStatus(Domain domain, Status status) {
     domain.setStatus(status);
-    return save(domain);
+    return saveStatus(domain);
   }
 
-  private Optional<Domain> save(Domain domain) {
-    return Optional.ofNullable(domainRepository.save(domain));
+  private Optional<Domain> saveSpecification(Domain domain) {
+    return Optional.ofNullable(domainRepository.saveSpecification(domain));
   }
 
+  private Optional<Domain> saveStatus(Domain domain) {
+    return Optional.ofNullable(domainRepository.saveStatus(domain));
+  }
+
+
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")
   public Optional<Domain> get(DomainKey key) {
     return domainView.get(key);

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/InfrastructureService.java

@@ -62,28 +62,32 @@ public Optional<Infrastructure> create(Infrastructure infrastructure) throws Val
     }
     infrastructureValidator.validateForCreate(infrastructure);
     infrastructure.setSpecification(handlerService.handleInsert(infrastructure));
-    return save(infrastructure);
+    return saveSpecification(infrastructure);
   }
 
   @PreAuthorize("hasPermission(#infrastructure, 'UPDATE')")
   public Optional<Infrastructure> update(Infrastructure infrastructure) throws ValidationException {
     val existing = infrastructureView.get(infrastructure.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + infrastructure.getKey().getName() + " because it doesn't exist");
     }
     infrastructureValidator.validateForUpdate(infrastructure, existing.get());
     infrastructure.setSpecification(handlerService.handleUpdate(infrastructure, existing.get()));
-    return save(infrastructure);
+    return saveSpecification(infrastructure);
   }
 
   @PreAuthorize("hasPermission(#infrastructure, 'UPDATE_STATUS')")
   public Optional<Infrastructure> updateStatus(Infrastructure infrastructure, Status status) {
     infrastructure.setStatus(status);
-    return save(infrastructure);
+    return saveStatus(infrastructure);
   }
 
-  private Optional<Infrastructure> save(Infrastructure infrastructure) {
-    return Optional.ofNullable(infrastructureRepository.save(infrastructure));
+  private Optional<Infrastructure> saveSpecification(Infrastructure infrastructure) {
+    return Optional.ofNullable(infrastructureRepository.saveSpecification(infrastructure));
+  }
+
+  private Optional<Infrastructure> saveStatus(Infrastructure infrastructure) {
+    return Optional.ofNullable(infrastructureRepository.saveStatus(infrastructure));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProcessBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2023 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -53,28 +53,32 @@ public Optional<ProcessBinding> create(ProcessBinding processBinding) throws Val
     }
     processBindingValidator.validateForCreate(processBinding);
     processBinding.setSpecification(handlerService.handleInsert(processBinding));
-    return save(processBinding);
+    return saveSpecification(processBinding);
   }
 
   @PreAuthorize("hasPermission(#processBinding, 'UPDATE')")
   public Optional<ProcessBinding> update(ProcessBinding processBinding) throws ValidationException {
     val existing = processBindingView.get(processBinding.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + processBinding.getKey() + " because it doesn't exist");
     }
     processBindingValidator.validateForUpdate(processBinding, existing.get());
     processBinding.setSpecification(handlerService.handleUpdate(processBinding, existing.get()));
-    return save(processBinding);
+    return saveSpecification(processBinding);
   }
 
   @PreAuthorize("hasPermission(#processBinding, 'UPDATE_STATUS')")
   public Optional<ProcessBinding> updateStatus(ProcessBinding processBinding, Status status) {
     processBinding.setStatus(status);
-    return save(processBinding);
+    return saveStatus(processBinding);
   }
 
-  private Optional<ProcessBinding> save(ProcessBinding processBinding) {
-    return Optional.ofNullable(processBindingRepository.save(processBinding));
+  private Optional<ProcessBinding> saveSpecification(ProcessBinding processBinding) {
+    return Optional.ofNullable(processBindingRepository.saveSpecification(processBinding));
+  }
+
+  private Optional<ProcessBinding> saveStatus(ProcessBinding processBinding) {
+    return Optional.ofNullable(processBindingRepository.saveStatus(processBinding));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProcessService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2023 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -74,7 +74,7 @@ public Optional<Process> create(Process process) throws ValidationException {
       process.getOutputs().forEach(output -> producerService.canCreateProducer(
         buildProducer(process, zoneKey, output)
       )));
-    return save(process);
+    return saveSpecification(process);
   }
 
   private Producer buildProducer(Process process, ZoneKey zoneKey, ProcessOutputStream output) {
@@ -94,7 +94,7 @@ private Producer buildProducer(Process process, ZoneKey zoneKey, ProcessOutputSt
   @PreAuthorize("hasPermission(#process, 'UPDATE')")
   public Optional<Process> update(Process process) throws ValidationException {
     val existing = processView.get(process.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + process.getKey().getName() + " because it doesn't exist");
     }
     processValidator.validateForUpdate(process, existing.get());
@@ -119,7 +119,7 @@ public Optional<Process> update(Process process) throws ValidationException {
       });
     });
 
-    return save(process);
+    return saveSpecification(process);
   }
 
   private Consumer buildConsumer(Process process, ZoneKey zoneKey, ProcessInputStream input) {
@@ -139,13 +139,18 @@ private Consumer buildConsumer(Process process, ZoneKey zoneKey, ProcessInputStr
   @PreAuthorize("hasPermission(#process, 'UPDATE_STATUS')")
   public Optional<Process> updateStatus(Process process, Status status) {
     process.setStatus(status);
-    return save(process);
+    return saveStatus(process);
   }
 
-  private Optional<Process> save(Process process) {
-    return Optional.ofNullable(processRepository.save(process));
+  private Optional<Process> saveSpecification(Process process) {
+    return Optional.ofNullable(processRepository.saveSpecification(process));
   }
 
+  private Optional<Process> saveStatus(Process process) {
+    return Optional.ofNullable(processRepository.saveStatus(process));
+  }
+
+
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")
   public Optional<Process> get(ProcessKey key) {
     return processView.get(key);

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProducerBindingService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2021 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -54,28 +54,32 @@ public Optional<ProducerBinding> create(ProducerBinding producerBinding) throws
     }
     producerBindingValidator.validateForCreate(producerBinding);
     producerBinding.setSpecification(handlerService.handleInsert(producerBinding));
-    return save(producerBinding);
+    return saveSpecification(producerBinding);
   }
 
   @PreAuthorize("hasPermission(#producerBinding, 'UPDATE')")
   public Optional<ProducerBinding> update(ProducerBinding producerBinding) throws ValidationException {
     val existing = producerBindingView.get(producerBinding.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + producerBinding.getKey() + " because it doesn't exist");
     }
     producerBindingValidator.validateForUpdate(producerBinding, existing.get());
     producerBinding.setSpecification(handlerService.handleUpdate(producerBinding, existing.get()));
-    return save(producerBinding);
+    return saveSpecification(producerBinding);
   }
 
   @PreAuthorize("hasPermission(#producerBinding, 'UPDATE_STATUS')")
   public Optional<ProducerBinding> updateStatus(ProducerBinding producerBinding, Status status) {
     producerBinding.setStatus(status);
-    return save(producerBinding);
+    return saveStatus(producerBinding);
   }
 
-  private Optional<ProducerBinding> save(ProducerBinding producerBinding) {
-    return Optional.ofNullable(producerBindingRepository.save(producerBinding));
+  private Optional<ProducerBinding> saveSpecification(ProducerBinding producerBinding) {
+    return Optional.ofNullable(producerBindingRepository.saveSpecification(producerBinding));
+  }
+
+  private Optional<ProducerBinding> saveStatus(ProducerBinding producerBinding) {
+    return Optional.ofNullable(producerBindingRepository.saveStatus(producerBinding));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")

core/src/main/java/com/expediagroup/streamplatform/streamregistry/core/services/ProducerService.java

@@ -1,5 +1,5 @@
 /**
- * Copyright (C) 2018-2022 Expedia, Inc.
+ * Copyright (C) 2018-2024 Expedia, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -56,28 +56,32 @@ public Optional<Producer> create(Producer producer) throws ValidationException {
     }
     producerValidator.validateForCreate(producer);
     producer.setSpecification(handlerService.handleInsert(producer));
-    return save(producer);
+    return saveSpecification(producer);
   }
 
   @PreAuthorize("hasPermission(#producer, 'UPDATE')")
   public Optional<Producer> update(Producer producer) throws ValidationException {
     val existing = producerView.get(producer.getKey());
-    if (!existing.isPresent()) {
+    if (existing.isEmpty()) {
       throw new ValidationException("Can't update " + producer.getKey().getName() + " because it doesn't exist");
     }
     producerValidator.validateForUpdate(producer, existing.get());
     producer.setSpecification(handlerService.handleUpdate(producer, existing.get()));
-    return save(producer);
+    return saveSpecification(producer);
   }
 
   @PreAuthorize("hasPermission(#producer, 'UPDATE_STATUS')")
   public Optional<Producer> updateStatus(Producer producer, Status status) {
     producer.setStatus(status);
-    return save(producer);
+    return saveStatus(producer);
   }
 
-  private Optional<Producer> save(Producer producer) {
-    return Optional.ofNullable(producerRepository.save(producer));
+  private Optional<Producer> saveSpecification(Producer producer) {
+    return Optional.ofNullable(producerRepository.saveSpecification(producer));
+  }
+
+  private Optional<Producer> saveStatus(Producer producer) {
+    return Optional.ofNullable(producerRepository.saveStatus(producer));
   }
 
   @PostAuthorize("returnObject.isPresent() ? hasPermission(returnObject, 'READ') : true")