Bug 36033341 - [36028299->23.09.2] PagedTopic EnsureChannelCountTask …

…may attempt to modify read-only XML (merge ce/main -> ce/23.09 104795) [git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v23.09/": change = 104796]
oracle · Nov 21, 2023 · 3d15e6a · 3d15e6a
1 parent 9d05775
commit 3d15e6a
Show file tree

Hide file tree

Showing 2 changed files with 111 additions and 17 deletions.
diff --git a/...il/daemon/queueProcessor/service/grid/partitionedService/partitionedCache/PagedTopic.java b/...il/daemon/queueProcessor/service/grid/partitionedService/partitionedCache/PagedTopic.java
@@ -50,6 +50,7 @@
 import com.tangosol.net.topic.TopicException;
 import com.tangosol.run.xml.SimpleElement;
 import com.tangosol.run.xml.XmlElement;
+import com.tangosol.run.xml.XmlValue;
 import com.tangosol.util.ExternalizableHelper;
 import com.tangosol.util.Filter;
 import com.tangosol.util.ListMap;
@@ -2613,18 +2614,12 @@ public String getTopicName()
         // From interface: java.lang.Runnable
         public void run()
             {
-            // import Component.Net.Cluster;
-            // import Component.Util.Daemon.QueueProcessor.Service.Grid$ServiceConfig$Map as com.tangosol.coherence.component.util.daemon.queueProcessor.service.Grid.ServiceConfig.Map;
-            // import com.oracle.coherence.common.base.Logger;
-            // import com.tangosol.internal.net.topic.impl.paged.PagedTopicCaches$Names as com.tangosol.internal.net.topic.impl.paged.PagedTopicCaches.Names;
-            // import java.util.concurrent.locks.ReentrantLock;
-
             PagedTopic service    = getService();
-            String  sTopic     = getTopicName();
-            int     cRequired  = getRequiredChannelCount();
-            int     cChannel   = getChannelCount();
-            int     cConfigMap = service.getChannelCountFromConfigMap(sTopic);
-            int     cActual;
+            String     sTopic     = getTopicName();
+            int        cRequired  = getRequiredChannelCount();
+            int        cChannel   = getChannelCount();
+            int        cConfigMap = service.getChannelCountFromConfigMap(sTopic);
+            int        cActual;
 
             if (cConfigMap == 0)
                 {
@@ -2657,11 +2652,16 @@ public void run()
                                 ((Cluster) service.getCluster()).suspendService(sServiceName, /*fResumeOnFailover*/ true);
                                 }
 
-                            String     sCacheName = com.tangosol.internal.net.topic.impl.paged.PagedTopicCaches.Names.CONTENT.cacheNameForTopicName(sTopic);
-                            com.tangosol.coherence.component.util.daemon.queueProcessor.service.Grid.ServiceConfig.Map      map        = service.getServiceConfigMap(); 
-                            XmlElement xmlElement = (XmlElement) map.get(sCacheName);
-
-                            xmlElement.getSafeAttribute("channels").setInt(cChannel);
+                            String            sCacheName   = PagedTopicCaches.Names.CONTENT.cacheNameForTopicName(sTopic);
+                            ServiceConfig.Map map          = service.getServiceConfigMap();
+                            XmlElement         xmlElement  = (XmlElement) map.get(sCacheName);
+                            XmlValue           xmlChannels = xmlElement.getAttribute("channels");
+
+                            if (xmlChannels == null)
+                                {
+                                xmlChannels = xmlElement.addAttribute("channels");
+                                }
+                            xmlChannels.setInt(cChannel);
                             map.put(sCacheName, xmlElement);
 
                             if (cConfigMap != cChannel)
@@ -2675,7 +2675,7 @@ public void run()
                             {
                             if (fSuspend)
                                 {
-                                ((Cluster) service.getCluster()).resumeService(sServiceName);
+                                service.getCluster().resumeService(sServiceName);
                                 }
                             }
                         }

diff --git a/prj/test/functional/topics/src/main/java/topics/EnsureChannelCountTaskTest.java b/prj/test/functional/topics/src/main/java/topics/EnsureChannelCountTaskTest.java
@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+ *
+ * Licensed under the Universal Permissive License v 1.0 as shown at
+ * https://oss.oracle.com/licenses/upl.
+ */
+
+package topics;
+
+import com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.PartitionedCache;
+import com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.partitionedCache.PagedTopic;
+import com.tangosol.coherence.component.util.daemon.queueProcessor.service.grid.partitionedService.partitionedCache.PagedTopic.EnsureChannelCountTask;
+
+import com.tangosol.internal.net.topic.impl.paged.PagedTopicCaches;
+import com.tangosol.run.xml.SimpleElement;
+import com.tangosol.run.xml.XmlElement;
+import org.junit.Test;
+
+import java.util.concurrent.locks.ReentrantLock;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+public class EnsureChannelCountTaskTest
+    {
+    @Test
+    public void shouldSetChannelCountIntoConfigMap()
+        {
+        String                 sTopicName = "test-topic";
+        String                 sCacheName = PagedTopicCaches.Names.CONTENT.cacheNameForTopicName(sTopicName);
+        ReentrantLock          lock       = new ReentrantLock();
+        XmlElement             xmlCache   = new SimpleElement();
+        EnsureChannelCountTask task       = new EnsureChannelCountTask();
+
+        task.setChannelCount(29);
+        task.setRequiredChannelCount(29);
+        task.setTopicName(sTopicName);
+
+        EnsureChannelCountTask stub = spy(task);
+
+        PartitionedCache.ServiceConfig.Map configMap = mock(PartitionedCache.ServiceConfig.Map.class);
+        PagedTopic                         service   = mock(PagedTopic.class);
+
+        when(stub.getService()).thenReturn(service);
+
+        when(service.getChannelCountFromConfigMap(sTopicName)).thenReturn(17);
+        when(service.getTopicStoreLock()).thenReturn(lock);
+        when(service.isSuspendedFully()).thenReturn(true);
+        when(service.getServiceConfigMap()).thenReturn(configMap);
+
+        when(configMap.get(sCacheName)).thenReturn(xmlCache);
+
+        stub.run();
+
+        assertThat(xmlCache.getSafeAttribute("channels").getInt(), is(29));
+        }
+
+    @Test
+    public void shouldUpdateChannelCountIntoConfigMap()
+        {
+        String                 sTopicName = "test-topic";
+        String                 sCacheName = PagedTopicCaches.Names.CONTENT.cacheNameForTopicName(sTopicName);
+        ReentrantLock          lock       = new ReentrantLock();
+        XmlElement             xmlCache   = new SimpleElement();
+        EnsureChannelCountTask task       = new EnsureChannelCountTask();
+
+        xmlCache.addElement("channels").setInt(17);
+
+        task.setChannelCount(29);
+        task.setRequiredChannelCount(29);
+        task.setTopicName(sTopicName);
+
+        EnsureChannelCountTask stub = spy(task);
+
+        PartitionedCache.ServiceConfig.Map configMap = mock(PartitionedCache.ServiceConfig.Map.class);
+        PagedTopic                         service   = mock(PagedTopic.class);
+
+        when(stub.getService()).thenReturn(service);
+
+        when(service.getChannelCountFromConfigMap(sTopicName)).thenReturn(17);
+        when(service.getTopicStoreLock()).thenReturn(lock);
+        when(service.isSuspendedFully()).thenReturn(true);
+        when(service.getServiceConfigMap()).thenReturn(configMap);
+
+        when(configMap.get(sCacheName)).thenReturn(xmlCache);
+
+        stub.run();
+
+        assertThat(xmlCache.getSafeAttribute("channels").getInt(), is(29));
+        }
+    }