Bug 36800725 - [36774493->24.03.1] Topics: New channels are not assigned to a subscriber when the count is increased

(merge ce/main -> ce/24.03 109931)

[git-p4: depot-paths = "//dev/coherence-ce/release/coherence-ce-v24.03/": change = 109932]

Author: thegridman

prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/util/daemon/queueProcessor/service/grid/partitionedService/partitionedCache/PagedTopic.java

@@ -1,6 +1,6 @@
 
 /*
- * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2024, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * https://oss.oracle.com/licenses/upl.
@@ -15,6 +15,7 @@
 import com.tangosol.internal.net.topic.impl.paged.PagedTopic;
 import com.tangosol.internal.net.topic.impl.paged.PagedTopicBackingMapManager;
 import com.tangosol.internal.net.topic.impl.paged.PagedTopicCaches;
+import com.tangosol.internal.net.topic.impl.paged.PagedTopicConfigMap;
 import com.tangosol.internal.net.topic.impl.paged.model.PagedTopicSubscription;
 import com.tangosol.net.PagedTopicService;
 import com.tangosol.net.TopicService;
@@ -320,6 +321,19 @@ public java.util.Set getSubscribers(String Param_1, com.tangosol.internal.net.to
         {
         return ((com.tangosol.net.PagedTopicService) getRunningCacheService()).getSubscribers(Param_1, Param_2);
         }
+
+    @Override
+    public boolean hasSubscribers(String sTopicName)
+        {
+        return ((com.tangosol.net.PagedTopicService) getRunningCacheService()).hasSubscribers(sTopicName);
+        }
+
+    @Override
+    public long getSubscriptionCount(String sTopicName)
+        {
+        return ((com.tangosol.net.PagedTopicService) getRunningCacheService()).getSubscriptionCount(sTopicName);
+        }
+
     @Override
     public com.tangosol.internal.net.topic.impl.paged.model.PagedTopicSubscription getSubscription(long Param_1)
         {

prj/coherence-core-components/src/main/java/com/tangosol/coherence/component/util/safeService/safeCacheService/safeDistributedCacheService/SafePagedTopicService.java

@@ -449,13 +449,14 @@ public List<NamedEventInterceptorBuilder> getEventInterceptorBuilders()
             }
 
         // add the subscriber expiry interceptor
-        NamedEventInterceptorBuilder builder = new NamedEventInterceptorBuilder();
-        builder.setOrder(Interceptor.Order.HIGH);
-        builder.setName("$SubscriberExpiry$" + getServiceName());
-        builder.setRegistrationBehavior(RegistrationBehavior.REPLACE);
-        builder.setCustomBuilder((resolver, loader, listParameters) -> new PagedTopicSubscriber.TimeoutInterceptor());
+        NamedEventInterceptorBuilder builderTimeout = new NamedEventInterceptorBuilder();
+        builderTimeout.setOrder(Interceptor.Order.HIGH);
+        builderTimeout.setName("$SubscriberExpiry$" + getServiceName());
+        builderTimeout.setRegistrationBehavior(RegistrationBehavior.REPLACE);
+        builderTimeout.setCustomBuilder((resolver, loader, listParameters) -> new PagedTopicSubscriber.TimeoutInterceptor());
+
+        list.add(builderTimeout);
 
-        list.add(builder);
         return list;
         }
 

prj/coherence-core/src/main/java/com/tangosol/coherence/config/scheme/PagedTopicScheme.java

@@ -322,26 +322,6 @@ public int getChannelCount()
         return f_topicService.getChannelCount(f_sTopicName);
         }
 
-    /**
-     * Return the default publisher channel count for this topic.
-     * <p>
-     * If the system property {@link Publisher#PROP_CHANNEL_COUNT} with a suffix of a dot
-     * followed by the topic name is set, that value will be used.
-     * <p>
-     * If the system property {@link Publisher#PROP_CHANNEL_COUNT} with no suffix is set,
-     * that value will be used.
-     * <p>
-     * If neither property is set the configured channel count will be used.
-     *
-     * @return the default publisher channel count for this topic
-     */
-    public int getPublisherChannelCount()
-        {
-        int cChannel = getDependencies().getConfiguredChannelCount();
-        cChannel = Config.getInteger(Publisher.PROP_CHANNEL_COUNT, cChannel);
-        return Config.getInteger(Publisher.PROP_CHANNEL_COUNT + "." + f_sTopicName, cChannel);
-        }
-
     /**
      * Return the set of NotificationKeys covering all partitions for the given notifier
      *

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicCaches.java

@@ -1,11 +1,13 @@
 /*
- * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2024, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * https://oss.oracle.com/licenses/upl.
  */
 package com.tangosol.internal.net.topic.impl.paged;
 
+import com.oracle.coherence.common.base.Logger;
+import com.tangosol.internal.net.topic.ChannelAllocationStrategy;
 import com.tangosol.internal.net.topic.impl.paged.model.PagedTopicSubscription;
 import com.tangosol.internal.net.topic.impl.paged.model.SubscriberGroupId;
 import com.tangosol.internal.net.topic.impl.paged.model.SubscriberId;
@@ -187,6 +189,69 @@ public static boolean hasSubscription(Map<?, ?> configMap, long lSubscriptionId,
         return false;
         }
 
+    /**
+     * Returns {@code true} if the specified topic has any registered subscriptions.
+     *
+     * @param configMap  the config map for the service
+     * @param sTopic     the name of the topic
+     *
+     * @return {@code true} if the specified topic has any registered subscriptions
+     */
+    public static boolean hasSubscriptions(Map<?, ?> configMap, String sTopic)
+        {
+        return configMap.keySet().stream()
+                .filter(key -> key instanceof PagedTopicSubscription.Key)
+                .map(PagedTopicSubscription.Key.class::cast)
+                .anyMatch(key -> key.getTopicName().equals(sTopic));
+        }
+
+    /**
+     * Returns the count of subscriptions for the specified topic.
+     *
+     * @param configMap  the config map for the service
+     * @param sTopic     the name of the topic
+     *
+     * @return the count of subscriptions for the specified topic
+     */
+    public static long getSubscriptionCount(Map<?, ?> configMap, String sTopic)
+        {
+        return configMap.keySet().stream()
+                .filter(key -> key instanceof PagedTopicSubscription.Key)
+                .map(PagedTopicSubscription.Key.class::cast)
+                .filter(key -> key.getTopicName().equals(sTopic))
+                .count();
+        }
+
+    /**
+     * Update the channel count and allocations for all subscriptions for a topic.
+     *
+     * @param configMap  the config map to update
+     * @param sTopic     the name of the topic
+     * @param cChannel   the new channel count
+     * @param strategy   the channel allocation strategy
+     */
+    @SuppressWarnings("unchecked")
+    public static void setChannelCount(Map<?, ?> configMap, String sTopic, int cChannel, ChannelAllocationStrategy strategy)
+        {
+        Set<PagedTopicSubscription.Key> setKey = configMap.keySet().stream()
+                .filter(key -> key instanceof PagedTopicSubscription.Key)
+                .map(PagedTopicSubscription.Key.class::cast)
+                .filter(key -> key.getTopicName().equals(sTopic))
+                .collect(Collectors.toSet());
+
+        Map<PagedTopicSubscription.Key, PagedTopicSubscription> mapSub = (Map<PagedTopicSubscription.Key, PagedTopicSubscription>) configMap;
+        for (PagedTopicSubscription.Key key : setKey)
+            {
+            PagedTopicSubscription subscription = (PagedTopicSubscription) configMap.get(key);
+            if (subscription.getChannelCount() < cChannel)
+                {
+                Logger.config("Updating channel count for subscription " + key.getGroupName() + " in topic " + sTopic + " to " + cChannel);
+                subscription.updateChannelAllocations(strategy, cChannel);
+                mapSub.put(key, subscription);
+                }
+            }
+        }
+
     /**
      * Remove configurations for the specified topic name.
      *

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicConfigMap.java

@@ -247,21 +247,12 @@ public OfferProcessor.Result offerToPageTail(BinaryEntry<Page.Key, Page> entry,
         int                    nChannel        = keyPage.getChannelId();
         long                   lPage           = keyPage.getPageId();
         PagedTopicDependencies configuration   = getDependencies();
-        int                    channelCount    = getChannelCount();
         int                    cbCapPage       = configuration.getPageCapacity();
         long                   cbCapServer     = configuration.getServerCapacity();
         List<Binary>           listElements    = processor.getElements();
         int                    nNotifyPostFull = processor.getNotifyPostFull();
         boolean                fSealPage       = processor.isSealPage();
 
-        if (nChannel >= channelCount)
-            {
-            // publisher tried to publish to a non-existent channel,
-            // the publisher probably has an incorrect channel count
-            throw new RequestIncompleteException("Invalid channel " + nChannel
-                    + ", channel count is " + channelCount + ", valid channels are 0.." + (channelCount - 1));
-            }
-
         if (cbCapServer > 0 && nNotifyPostFull != 0)
             {
             // check if we own a large enough number of partitions that we would exceed the server capacity before
@@ -453,13 +444,16 @@ protected Page ensurePage(int nChannel, BinaryEntry<Page.Key, Page> entry)
         usage.setPartitionTail(lPage);
         usage.setPartitionMax(lPage); // unlike tail this is never reset to NULL_PAGE
 
+        boolean fFirstPage;
         if (lTailPrev == Page.NULL_PAGE)
             {
             // partition was empty, our new tail is also our head
             usage.setPartitionHead(lPage);
+            fFirstPage = true;
             }
         else
             {
+            fFirstPage = false;
             // attach old tail to new tail
             page.incrementReferenceCount(); // ref from old tail to new page
 
@@ -471,7 +465,20 @@ protected Page ensurePage(int nChannel, BinaryEntry<Page.Key, Page> entry)
             }
 
         // attach on behalf of waiting subscribers, they will have to find this page on their own
-        page.adjustReferenceCount(usage.resetWaitingSubscriberCount());
+        int c = usage.resetWaitingSubscriberCount();
+        if (fFirstPage && c == 0)
+            {
+            // The Usage may have zero subscriptions if a Publisher has recently increased the channel count
+            // In this case we check the service to see whether the topic has subscriptions
+            String sCache        = entry.getBackingMapContext().getCacheName();
+            String sTopic        = PagedTopicCaches.Names.getTopicName(sCache);
+            long   cSubscription = f_service.getSubscriptionCount(sTopic);
+            if (cSubscription > 0L)
+                {
+                c = 1;//(int) cSubscription;
+                }
+            }
+        page.adjustReferenceCount(c);
 
         long nTime = getClusterTime();
         page.setTimestampHead(nTime);

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicPartition.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2024, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * https://oss.oracle.com/licenses/upl.
@@ -116,7 +116,7 @@ public PagedTopicPublisher(NamedTopic<V> topic, PagedTopicCaches pagedTopicCache
 
         ChannelCount channelCount = options.get(ChannelCount.class, ChannelCount.USE_CONFIGURED);
         int          cChannel     = channelCount.isUseConfigured()
-                                            ? pagedTopicCaches.getPublisherChannelCount()
+                                            ? topic.getChannelCount()
                                             : channelCount.getChannelCount();
 
         long cbBatch  = m_caches.getDependencies().getMaxBatchSizeBytes();

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicPublisher.java

@@ -2381,9 +2381,14 @@ private void updateChannelOwnership(SortedSet<Integer> setChannel, boolean fLost
             PagedTopicChannel[] aExistingChannel = m_aChannel;
             if (nMaxChannel >= aExistingChannel.length)
                 {
-                int cChannel = nMaxChannel + 1;
-                // this subscriber has fewer channels than the server so needs to be resized
-                m_aChannel = initializeChannels(m_caches, cChannel, f_subscriberGroupId, aExistingChannel);
+                // This subscriber has fewer channels than the server so needs to be resized
+                // We disconnect as the subscription may not be properly initialized for
+                // the new channel count if this has happened due to a rolling upgrade
+                // from an earlier buggy topics version
+                Logger.finer(() -> String.format("Disconnecting subscriber %d on topic %s due to increase in channel count from %d to %d",
+                        f_id.getId(), f_topic.getName(), aExistingChannel.length, nMaxChannel));
+                disconnectInternal(true);
+                return;
                 }
 
             if (!Arrays.equals(m_aChannelOwned, aChannel))
@@ -2401,7 +2406,7 @@ private void updateChannelOwnership(SortedSet<Integer> setChannel, boolean fLost
                     }
                 setRevoked = Collections.unmodifiableSet(setRevoked);
 
-                Set<Integer> setAssigned = Collections.unmodifiableSet(new HashSet<>(setChannel));
+                Set<Integer> setAssigned = Set.copyOf(setChannel);
 
                 Logger.finest(String.format("Subscriber %d (name=%s) channel allocation changed, assigned=%s added=%s revoked=%s",
                                             f_id.getId(), f_sIdentifyingName, setAssigned, setAdded, setRevoked));

prj/coherence-core/src/main/java/com/tangosol/internal/net/topic/impl/paged/PagedTopicSubscriber.java

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2000, 2023, Oracle and/or its affiliates.
+ * Copyright (c) 2000, 2024, Oracle and/or its affiliates.
  *
  * Licensed under the Universal Permissive License v 1.0 as shown at
  * https://oss.oracle.com/licenses/upl.
@@ -9,6 +9,7 @@
 
 import com.tangosol.internal.net.topic.impl.paged.PagedTopicBackingMapManager;
 
+import com.tangosol.internal.net.topic.impl.paged.PagedTopicConfigMap;
 import com.tangosol.internal.net.topic.impl.paged.model.PagedTopicSubscription;
 import com.tangosol.internal.net.topic.impl.paged.model.SubscriberGroupId;
 import com.tangosol.internal.net.topic.impl.paged.model.SubscriberId;
@@ -195,6 +196,24 @@ long ensureSubscription(String sTopicName, SubscriberGroupId groupId, Subscriber
      */
     Set<SubscriberId> getSubscribers(String sTopicName, SubscriberGroupId groupId);
 
+    /**
+     * Returns {@code true} if the specified topic has subscribers or subscriber groups.
+     *
+     * @param sTopicName the name of the topic
+     *
+     * @return {@code true} if the specified topic has subscribers or subscriber groups
+     */
+    boolean hasSubscribers(String sTopicName);
+
+    /**
+     * Returns the count of subscriptions for the specified topic.
+     *
+     * @param sTopicName the name of the topic
+     *
+     * @return the count of subscriptions for the specified topic
+     */
+    long getSubscriptionCount(String sTopicName);
+
     /**
      * Add a listener that will be notified when changes are made to topic subscriptions.
      *