Repository: apache/streampipes
Pull request #3073 (2024-07-30 06:57:24)

fix(#3072): nullpointer in datalakemeasureresourcev4getdatalakeinfos
<!--
  ~ Licensed to the Apache Software Foundation (ASF) under one or more
  ~ contributor license agreements.  See the NOTICE file distributed with
  ~ this work for additional information regarding copyright ownership.
  ~ The ASF licenses this file to You under the Apache License, Version 2.0
  ~ (the "License"); you may not use this file except in compliance with
  ~ the License.  You may obtain a copy of the License at
  ~
  ~    http://www.apache.org/licenses/LICENSE-2.0
  ~
  ~ Unless required by applicable law or agreed to in writing, software
  ~ distributed under the License is distributed on an "AS IS" BASIS,
  ~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  ~ See the License for the specific language governing permissions and
  ~ limitations under the License.
  ~
  -->

  <!--
Thanks for contributing! Here are some tips you can follow to help us incorporate your contribution quickly and easily:
1. If this is your first time, please read our contributor guidelines:
    - https://streampipes.apache.org/community/get-involved/
    - https://cwiki.apache.org/confluence/display/STREAMPIPES/Getting+Started
2. Make sure the PR title is formatted like: `[#<GitHub issue id>] PR title ...`
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., `[WIP][#<GitHub issue id>] PR title ...`.
4. Please write your PR title to summarize what this PR proposes/fixes.
5. Link the PR to the corresponding GitHub issue (if present) in the `Development` section in the right menu bar.
6. Be sure to keep the PR description updated to reflect all changes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
8. Make sure tests pass via `mvn clean install`.
9. (Optional) If the contribution is large, please file an Apache ICLA
    - http://apache.org/licenses/icla.pdf
-->

### Purpose
<!--
Please clarify what changes you are proposing and describe how those changes will address the issue.
Furthermore, describe potential consequences the changes might have.
-->
- Fixes a NullPointerException when the provided measurement name does not exist
- Adds unit tests to cover this scenario
- Renames the REST API endpoint for retrieving event counts of measurements for better clarity
- Updates the documentation to reflect the functionality in the REST endpoint

### Old API:
```
GET /api/v4/datalake/measure
```
**Parameters:**
- `filter` (optional): List of measurement names

### New API:
```
GET /api/v4/datalake/measure/count
```
**Parameters:**
- `measurementNames`: List of measurement names to return the count

### Remarks
<!--
Is there anything left we need to pay attention on?
Are there some references that might be important? E.g. links to Confluence, or discussions
on the mailing list or GitHub.
-->
PR introduces (a) breaking change(s): yes

PR introduces (a) deprecation(s): no


Sylvain2703: Looks great to me, thanks for your responsiveness 😄

Patch:
diff --git a/docker-compose.yml b/docker-compose.yml
index c11d7036fe..e993154964 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -20,19 +20,16 @@
 # provided on https://streampipes.apache.org/download
 #
 
-
 version: "3.8"
 
 # global logging
-x-logging:
-  &default-logging
+x-logging: &default-logging
   options:
-    max-size: '12m'
-    max-file: '5'
+    max-size: "12m"
+    max-file: "5"
   driver: json-file
 
 services:
-
   #### apache/streampipes
   backend:
     build:
@@ -135,7 +132,6 @@ services:
     networks:
       spnet:
 
-
   mosquitto:
     image: eclipse-mosquitto:1.5.4
     logging: *default-logging
@@ -146,7 +142,6 @@ services:
     image: mcr.microsoft.com/iotedge/opc-plc
     logging: *default-logging
     command: --ut=true
-    logging: *default-logging
     networks:
       spnet:
 
diff --git a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
index 0c1026a627..844ce5b573 100644
--- a/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
+++ b/streampipes-data-explorer/src/main/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounter.java
@@ -30,6 +30,7 @@ import org.slf4j.LoggerFactory;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
@@ -53,6 +54,7 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
     // create async futures so that count queries can be executed parallel
     Map<String, CompletableFuture<Integer>> countQueriesFutures = measurementNames.stream()
         .map(this::getMeasure)
+        .filter(Objects::nonNull)
         .collect(Collectors.toMap(
             DataLakeMeasure::getMeasureName,
             this::createQueryAsAsyncFuture)
@@ -97,10 +99,6 @@ public abstract class DataLakeMeasurementCounter implements IDataLakeMeasurement
     return resultPerMeasure;
   }
 
-  private Integer getQueryResult(Map<String, CompletableFuture<Integer>> queryFuture) {
-    return null;
-  }
-
   /**
    * Retrieves the runtime name of the first measurement property from the event schema of the given DataLakeMeasure.
    *
diff --git a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
index 266bca7b27..2e0f902745 100644
--- a/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
+++ b/streampipes-platform-services/src/main/java/org/apache/streampipes/ps/DataLakeMeasureResourceV4.java
@@ -23,6 +23,8 @@ import org.apache.streampipes.dataexplorer.management.DataExplorerDispatcher;
 import org.apache.streampipes.model.datalake.DataLakeMeasure;
 import org.apache.streampipes.rest.core.base.impl.AbstractAuthGuardedRestResource;
 
+import io.swagger.v3.oas.annotations.Operation;
+import io.swagger.v3.oas.annotations.Parameter;
 import org.springframework.http.MediaType;
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.DeleteMapping;
@@ -47,7 +49,7 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource {
 
   public DataLakeMeasureResourceV4() {
     this.dataLakeMeasureManagement = new DataExplorerDispatcher().getDataExplorerManager()
-        .getSchemaManagement();
+                                                                 .getSchemaManagement();
   }
 
   @PostMapping(
@@ -59,16 +61,32 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource {
     return ok(result);
   }
 
-  @GetMapping(produces = MediaType.APPLICATION_JSON_VALUE)
-  public ResponseEntity<Map<String, Integer>> getDataLakeInfos(
-      @RequestParam(value = "filter", required = false) List<String> measurementNames) {
+  /**
+   * Handles HTTP GET requests to retrieve the entry counts of specified measurements.
+   *
+   * @param measurementNames A list of measurement names to return the count.
+   * @return A ResponseEntity containing a map of measurement names and their corresponding entry counts.
+   */
+  @Operation(
+      summary = "Retrieve measurement counts",
+      description = "Retrieves the entry counts for the specified measurements from the data lake.")
+  @GetMapping(
+      path = "/count",
+      produces = MediaType.APPLICATION_JSON_VALUE)
+  public ResponseEntity<Map<String, Integer>> getEntryCountsOfMeasurments(
+      @Parameter(description = "A list of measurement names to return the count.")
+      @RequestParam(value = "measurementNames")
+      List<String> measurementNames
+  ) {
     var allMeasurements = this.dataLakeMeasureManagement.getAllMeasurements();
-    return ok(new DataExplorerDispatcher().getDataExplorerManager()
+    var result = new DataExplorerDispatcher()
+        .getDataExplorerManager()
         .getMeasurementCounter(
             allMeasurements,
             measurementNames
         )
-        .countMeasurementSizes());
+        .countMeasurementSizes();
+    return ok(result);
   }
 
   @GetMapping(path = "{id}", produces = MediaType.APPLICATION_JSON_VALUE)
@@ -82,8 +100,10 @@ public class DataLakeMeasureResourceV4 extends AbstractAuthGuardedRestResource {
   }
 
   @PutMapping(path = "{id}", consumes = MediaType.APPLICATION_JSON_VALUE)
-  public ResponseEntity<?> updateDataLakeMeasure(@PathVariable("id") String elementId,
-                                                 @RequestBody DataLakeMeasure measure) {
+  public ResponseEntity<?> updateDataLakeMeasure(
+      @PathVariable("id") String elementId,
+      @RequestBody DataLakeMeasure measure
+  ) {
     if (elementId.equals(measure.getElementId())) {
       try {
         this.dataLakeMeasureManagement.updateMeasurement(measure);
diff --git a/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts b/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
index aa2e62c1b2..a8e8c8928f 100644
--- a/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
+++ b/ui/projects/streampipes/platform-services/src/lib/apis/datalake-rest.service.ts
@@ -46,11 +46,11 @@ export class DatalakeRestService {
     }
 
     getMeasurementEntryCounts(
-        measurementNameFilter: string[],
+        measurementNames: string[],
     ): Observable<Record<string, number>> {
         return this.http
-            .get(`${this.dataLakeMeasureUrl}`, {
-                params: { filter: measurementNameFilter },
+            .get(`${this.dataLakeMeasureUrl}/count`, {
+                params: { measurementNames: measurementNames },
             })
             .pipe(map(r => r as Record<string, number>));
     }


Test patch:
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTest.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTest.java
new file mode 100644
index 0000000000..7f73f6685e
--- /dev/null
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.query;
+
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+
+public class DataLakeMeasurementCounterTest {
+
+  private static final String MEASURE_1_NAME = "measure1";
+  private static final String MEASURE_2_NAME = "measure2";
+  private static final String UNKNOWN_MEASURE_NAME = "unknownMeasureName";
+
+  private DataLakeMeasurementCounterTestImpl counter;
+
+  private List<DataLakeMeasure> allMeasurements;
+
+  @BeforeEach
+  void setUp() {
+    var measure1 = mock(DataLakeMeasure.class);
+    var measure2 = mock(DataLakeMeasure.class);
+    when(measure1.getMeasureName()).thenReturn(MEASURE_1_NAME);
+    when(measure2.getMeasureName()).thenReturn(MEASURE_2_NAME);
+
+    allMeasurements = List.of(measure1, measure2);
+  }
+
+  @Test
+  void countMeasurementSizes_WithTwoEntries() {
+    var measurementNames = List.of(MEASURE_1_NAME, MEASURE_2_NAME);
+
+    counter = new DataLakeMeasurementCounterTestImpl(allMeasurements, measurementNames);
+    var result = counter.countMeasurementSizes();
+    assertEquals(2, result.size());
+    assertEquals(1, result.get(MEASURE_1_NAME));
+    assertEquals(1, result.get(MEASURE_2_NAME));
+  }
+
+  @Test
+  void countMeasurementSizes_WithNotExistingMeasureName() {
+    var measurementNames = List.of(UNKNOWN_MEASURE_NAME);
+    counter = new DataLakeMeasurementCounterTestImpl(allMeasurements, measurementNames);
+
+    var result = counter.countMeasurementSizes();
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  void countMeasurementSizes_WithEmptyMeasureNames() {
+    counter = new DataLakeMeasurementCounterTestImpl(allMeasurements, List.of());
+
+    var result = counter.countMeasurementSizes();
+    assertEquals(0, result.size());
+  }
+
+  @Test
+  void countMeasurementSizes_WithEmptyMeasurementsAndMeasureNames() {
+    counter = new DataLakeMeasurementCounterTestImpl(List.of(), List.of());
+
+    var result = counter.countMeasurementSizes();
+    assertEquals(0, result.size());
+  }
+}
\ No newline at end of file
diff --git a/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTestImpl.java b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTestImpl.java
new file mode 100644
index 0000000000..4642c5f227
--- /dev/null
+++ b/streampipes-data-explorer/src/test/java/org/apache/streampipes/dataexplorer/query/DataLakeMeasurementCounterTestImpl.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.streampipes.dataexplorer.query;
+
+import org.apache.streampipes.model.datalake.DataLakeMeasure;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+public class DataLakeMeasurementCounterTestImpl extends DataLakeMeasurementCounter {
+  public DataLakeMeasurementCounterTestImpl(
+      List<DataLakeMeasure> allMeasurements, List<String> measurementNames
+  ) {
+    super(allMeasurements, measurementNames);
+  }
+
+  @Override
+  protected CompletableFuture<Integer> createQueryAsAsyncFuture(DataLakeMeasure measure) {
+    // Mock implementation for testing
+    return CompletableFuture.completedFuture(1);
+  }
+}
\ No newline at end of file