ISSUE-212 - Fix bug in persisting view properties while connecting to stream (#213)

* [ISSUE-212] Fix bug in accidentally persisting View partitions on stream connect

* update javadoc

Author: Crim

CHANGELOG.md

@@ -5,6 +5,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 ## 2.5.1 (05/19/2020)
 - [ISSUE-209](https://github.com/SourceLabOrg/kafka-webview/issues/209) Expose HealthCheck and App Info endpoints without requiring authentication.
 - [ISSUE-161](https://github.com/SourceLabOrg/kafka-webview/issues/161) Add dedicated 'Apply' and 'Reset' button to Partition and Record Filters.
+- [ISSUE-212](https://github.com/SourceLabOrg/kafka-webview/issues/212) Bugfix for partition filters being persisted when toggled on from `Stream` page.
 
 #### Internal Dependency Updates
 -[PR-198](https://github.com/SourceLabOrg/kafka-webview/pull/198) Upgrade from SpringBoot 2.1.9 to 2.1.14.

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/stream/StreamController.java

@@ -44,13 +44,13 @@
 import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
 import org.springframework.security.core.Authentication;
 import org.springframework.stereotype.Controller;
+import org.springframework.transaction.annotation.Transactional;
 import org.springframework.ui.Model;
 import org.springframework.web.bind.annotation.PathVariable;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.servlet.mvc.support.RedirectAttributes;
 
-import javax.transaction.Transactional;
 import java.util.List;
 import java.util.Optional;
 
@@ -118,7 +118,7 @@ public String stream(
      * Serves websocket requests, requesting to start a stream on the given view.
      */
     @MessageMapping("/consume/{viewId}")
-    @Transactional
+    @Transactional(readOnly = true)
     public String newConsumer(
         @DestinationVariable final Long viewId,
         final ConsumeRequest consumeRequest,
@@ -153,7 +153,7 @@ public String newConsumer(
      * Serves web socket requests, requesting to pause a consumer.
      */
     @MessageMapping("/pause/{viewId}")
-    @Transactional
+    @Transactional(readOnly = true)
     public String pauseConsumer(
         @DestinationVariable final Long viewId,
         final SimpMessageHeaderAccessor headerAccessor) {
@@ -169,7 +169,7 @@ public String pauseConsumer(
      * Serves web socket requests, requesting to resume a consumer.
      */
     @MessageMapping("/resume/{viewId}")
-    @Transactional
+    @Transactional(readOnly = true)
     public String resumeConsumer(
         @DestinationVariable final Long viewId,
         final SimpMessageHeaderAccessor headerAccessor) {

kafka-webview-ui/src/main/resources/templates/stream/index.html

@@ -217,9 +217,6 @@ <h4 class="alert-heading">Stream Disconnected</h4>
                 // Merge parameters with start position parameters.
                 jQuery.extend(params, SocketDetails.getStartPositionParams());
 
-                // TODO remove.
-                console.log("Params: " + JSON.stringify(params));
-
                 // Send request down websocket.
                 SocketDetails.stompClient.send(SocketDetails.getSubscribeUrl(), {}, JSON.stringify(params));
             },

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/controller/stream/StreamControllerTest.java

@@ -24,18 +24,33 @@
 
 package org.sourcelab.kafka.webview.ui.controller.stream;
 
+import com.salesforce.kafka.test.junit4.SharedKafkaTestResource;
+import org.junit.Assert;
+import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.sourcelab.kafka.webview.ui.controller.AbstractMvcTest;
+import org.sourcelab.kafka.webview.ui.controller.api.requests.ConsumeRequest;
+import org.sourcelab.kafka.webview.ui.manager.socket.WebSocketConsumersManager;
+import org.sourcelab.kafka.webview.ui.model.Cluster;
 import org.sourcelab.kafka.webview.ui.model.View;
+import org.sourcelab.kafka.webview.ui.repository.ViewRepository;
+import org.sourcelab.kafka.webview.ui.tools.ClusterTestTools;
 import org.sourcelab.kafka.webview.ui.tools.ViewTestTools;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.autoconfigure.web.servlet.AutoConfigureMockMvc;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.messaging.simp.SimpMessageHeaderAccessor;
+import org.springframework.security.core.Authentication;
 import org.springframework.test.context.junit4.SpringRunner;
 import org.springframework.transaction.annotation.Transactional;
 
+import java.util.ArrayList;
+
 import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 import static org.springframework.security.test.web.servlet.request.SecurityMockMvcRequestPostProcessors.user;
 import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
 import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
@@ -45,10 +60,24 @@
 @SpringBootTest
 @AutoConfigureMockMvc
 public class StreamControllerTest extends AbstractMvcTest {
+    @ClassRule
+    public static SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource();
 
     @Autowired
     private ViewTestTools viewTestTools;
 
+    @Autowired
+    private ViewRepository viewRepository;
+
+    @Autowired
+    private ClusterTestTools clusterTestTools;
+
+    @Autowired
+    private StreamController streamController;
+
+    @Autowired
+    private WebSocketConsumersManager webSocketConsumersManager;
+
     /**
      * Ensure authentication is required.
      */
@@ -68,6 +97,7 @@ public void testUrlsRequireAuthentication() throws Exception {
     @Test
     @Transactional
     public void smokeTestStream() throws Exception {
+
         final View view = viewTestTools.createView("TestView");
 
         // Hit the stream page for specified view.
@@ -82,4 +112,68 @@ public void smokeTestStream() throws Exception {
             .andExpect(content().string(containsString("Switch to View")))
             .andExpect(content().string(containsString("href=\"/view/" + view.getId() + "\"")));
     }
+
+    /**
+     * Test starting a new websocket consumer.
+     * Verifies bug uncovered in ISSUE-212 https://github.com/SourceLabOrg/kafka-webview/issues/212
+     */
+    @Test
+    public void shouldReceiveAMessageFromTheServer() throws Exception {
+        final String expectedSessionId = "MYSESSIONID";
+        final String topicName = "TestTopic" + System.currentTimeMillis();
+
+        // Sanity test, no active consumers
+        Assert.assertEquals("Should have no active consumers", 0, webSocketConsumersManager.countActiveConsumers());
+
+        // Create a topic
+        sharedKafkaTestResource
+            .getKafkaTestUtils()
+            .createTopic(topicName, 2, (short) 1);
+
+        // Create cluster
+        final Cluster cluster = clusterTestTools
+            .createCluster("TestCluster", sharedKafkaTestResource.getKafkaConnectString());
+
+        // Create view
+        final View view = viewTestTools
+            .createViewWithCluster("TestView", cluster);
+
+        // Sanity test, no enforced partitions
+        assertEquals("Partitions Property should be empty string", "", view.getPartitions());
+
+        final ConsumeRequest consumeRequest = new ConsumeRequest();
+        consumeRequest.setAction("head");
+        consumeRequest.setPartitions("0,1");
+        consumeRequest.setFilters(new ArrayList<>());
+        consumeRequest.setResultsPerPartition(10);
+
+        final SimpMessageHeaderAccessor mockHeaderAccessor = mock(SimpMessageHeaderAccessor.class);
+        final Authentication mockPrincipal = mock(Authentication.class);
+        when(mockHeaderAccessor.getUser())
+            .thenReturn(mockPrincipal);
+        when(mockPrincipal.getPrincipal())
+            .thenReturn(nonAdminUserDetails);
+        when(mockHeaderAccessor.getSessionId())
+            .thenReturn(expectedSessionId);
+
+        try {
+            final String result = streamController.newConsumer(
+                view.getId(),
+                consumeRequest,
+                mockHeaderAccessor
+            );
+            assertEquals("Should return success response", "{success: true}", result);
+
+            // Verify consumer stood up
+            assertEquals("Should have one active consumer", 1, webSocketConsumersManager.countActiveConsumers());
+
+            // Lets refresh the view entity and verify the partitions field is still empty.
+            // Validates ISSUE-212
+            final View updatedView = viewRepository.findById(view.getId()).get();
+            assertEquals("Partitions Property should be empty string", "", updatedView.getPartitions());
+        } finally {
+            // Cleanup, disconnect websocket consumers
+            webSocketConsumersManager.removeConsumersForSessionId(expectedSessionId);
+        }
+    }
 }

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/tools/ViewTestTools.java

@@ -31,6 +31,7 @@
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.persistence.EntityManager;
 import java.sql.Timestamp;
 
 /**
@@ -47,15 +48,18 @@ public class ViewTestTools {
     private final ViewRepository viewRepository;
     private final ClusterTestTools clusterTestTools;
     private final MessageFormatTestTools messageFormatTestTools;
+    private final EntityManager entityManager;
 
     @Autowired
     public ViewTestTools(
         final ViewRepository viewRepository,
         final ClusterTestTools clusterTestTools,
-        final MessageFormatTestTools messageFormatTestTools) {
+        final MessageFormatTestTools messageFormatTestTools,
+        final EntityManager entityManager) {
         this.viewRepository = viewRepository;
         this.clusterTestTools = clusterTestTools;
         this.messageFormatTestTools = messageFormatTestTools;
+        this.entityManager = entityManager;
     }
 
     public View createView(final String name) {
@@ -118,4 +122,13 @@ public void save(final View view) {
         viewRepository.save(view);
     }
 
+    /**
+     * Refresh/reload entity from Database.
+     * @param view View instance to reload.
+     * @return View instance.
+     */
+    public View reload(final View view) {
+        entityManager.refresh(view);
+        return view;
+    }
 }