Wip/wan tx grouping module 1

boms/geode-all-bom/src/test/resources/expected-pom.xml

@@ -912,6 +912,12 @@
         <version>${version}</version>
         <scope>compile</scope>
       </dependency>
+      <dependency>
+        <groupId>org.apache.geode</groupId>
+        <artifactId>geode-wan-txgrouping</artifactId>
+        <version>${version}</version>
+        <scope>compile</scope>
+      </dependency>
       <dependency>
         <groupId>org.apache.geode</groupId>
         <artifactId>geode-web</artifactId>

buildSrc/src/main/resources/japicmp_exceptions.json

@@ -2,5 +2,11 @@
   "Class org.apache.geode.management.builder.GeodeClusterManagementServiceBuilder": "Moved internal class to fix split packages between geode-core and geode-management",
   "Class org.apache.geode.management.api.ClusterManagementOperation": "Fixed missing @Experimental annotation",
   "Method org.apache.geode.management.api.ClusterManagementOperation.getEndpoint()": "Fixed missing @Experimental annotation",
-  "Method org.apache.geode.management.api.ClusterManagementOperation.getOperator()": "Fixed missing @Experimental annotation"
+  "Method org.apache.geode.management.api.ClusterManagementOperation.getOperator()": "Fixed missing @Experimental annotation",
+  "Class org.apache.geode.cache.wan.GatewaySender":"Added to support new types of gatewaysenders",
+  "Method org.apache.geode.cache.wan.GatewaySender.getType()":"Added to support new types of gatewaysenders",
+  "Class org.apache.geode.cache.wan.GatewaySenderFactory":"Added to support new types of gatewaysenders",
+  "Method org.apache.geode.cache.wan.GatewaySenderFactory.setType(java.lang.String)":"Added to support new types of gatewaysenders",
+  "Class org.apache.geode.management.GatewaySenderMXBean":"Added to support new types of gatewaysenders",
+  "Method org.apache.geode.management.GatewaySenderMXBean.getType()":"Added to support new types of gatewaysenders"
 }

geode-core/src/integrationTest/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImplJUnitTest.java

@@ -24,7 +24,7 @@
 
 import org.apache.geode.cache.CacheFactory;
 import org.apache.geode.internal.cache.InternalCache;
-import org.apache.geode.internal.cache.wan.GatewaySenderAttributes;
+import org.apache.geode.internal.cache.wan.GatewaySenderAttributesImpl;
 import org.apache.geode.test.junit.categories.AEQTest;
 
 @Category({AEQTest.class})
@@ -55,7 +55,7 @@ public void tearDown() {
   @Test
   public void testStopClearsStats() {
 
-    GatewaySenderAttributes attrs = new GatewaySenderAttributes();
+    GatewaySenderAttributesImpl attrs = new GatewaySenderAttributesImpl();
     String tempId = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
     attrs.setId(tempId);
     SerialAsyncEventQueueImpl queue = new SerialAsyncEventQueueImpl(cache,
@@ -81,7 +81,7 @@ public void testStopClearsStats() {
 
   @Test
   public void testStopStart() {
-    GatewaySenderAttributes attrs = new GatewaySenderAttributes();
+    GatewaySenderAttributesImpl attrs = new GatewaySenderAttributesImpl();
     String tempId = AsyncEventQueueImpl.ASYNC_EVENT_QUEUE_PREFIX + "id";
     attrs.setId(tempId);
     SerialAsyncEventQueueImpl queue = new SerialAsyncEventQueueImpl(cache,

geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/AsyncEventQueueFactoryImpl.java

@@ -28,7 +28,7 @@
 import org.apache.geode.cache.wan.GatewaySender.OrderPolicy;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.wan.AsyncEventQueueConfigurationException;
-import org.apache.geode.internal.cache.wan.GatewaySenderAttributes;
+import org.apache.geode.internal.cache.wan.GatewaySenderAttributesImpl;
 import org.apache.geode.internal.cache.wan.InternalGatewaySender;
 import org.apache.geode.internal.cache.xmlcache.AsyncEventQueueCreation;
 import org.apache.geode.internal.cache.xmlcache.CacheCreation;
@@ -53,13 +53,14 @@ public class AsyncEventQueueFactoryImpl implements AsyncEventQueueFactory {
    * Used internally to pass the attributes from this factory to the real GatewaySender it is
    * creating.
    */
-  private final GatewaySenderAttributes gatewaySenderAttributes;
+  private final GatewaySenderAttributesImpl gatewaySenderAttributes;
 
   public AsyncEventQueueFactoryImpl(InternalCache cache) {
-    this(cache, new GatewaySenderAttributes(), DEFAULT_BATCH_TIME_INTERVAL);
+    this(cache, new GatewaySenderAttributesImpl(), DEFAULT_BATCH_TIME_INTERVAL);
   }
 
-  AsyncEventQueueFactoryImpl(InternalCache cache, GatewaySenderAttributes gatewaySenderAttributes,
+  AsyncEventQueueFactoryImpl(InternalCache cache,
+      GatewaySenderAttributesImpl gatewaySenderAttributes,
       int batchTimeInterval) {
     this.cache = cache;
     this.gatewaySenderAttributes = gatewaySenderAttributes;

geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/ParallelAsyncEventQueueImpl.java

@@ -210,6 +210,11 @@ public void setModifiedEventId(EntryEventImpl clonedEvent) {
     clonedEvent.setEventId(newEventId);
   }
 
+  @Override
+  public String getType() {
+    return "ParallelAsyncEventQueue";
+  }
+
   private ThreadsMonitoring getThreadMonitorObj() {
     DistributionManager distributionManager = cache.getDistributionManager();
     if (distributionManager != null) {
@@ -218,4 +223,5 @@ private ThreadsMonitoring getThreadMonitorObj() {
       return null;
     }
   }
+
 }

geode-core/src/main/java/org/apache/geode/cache/asyncqueue/internal/SerialAsyncEventQueueImpl.java

@@ -264,6 +264,11 @@ public void setModifiedEventId(EntryEventImpl clonedEvent) {
     clonedEvent.setEventId(newEventId);
   }
 
+  @Override
+  public String getType() {
+    return "SerialAsyncEventQueue";
+  }
+
   private ThreadsMonitoring getThreadMonitorObj() {
     DistributionManager distributionManager = cache.getDistributionManager();
     if (distributionManager != null) {

geode-core/src/main/java/org/apache/geode/cache/configuration/CacheConfig.java

@@ -2588,6 +2588,7 @@ public void setOverflowDirectory(String value) {
    *       <attribute name="id" use="required" type="{http://www.w3.org/2001/XMLSchema}string" />
    *       <attribute name="remote-distributed-system-id" use="required" type="{http://www.w3.org/2001/XMLSchema}string" />
    *       <attribute name="parallel" type="{http://www.w3.org/2001/XMLSchema}boolean" />
+   *       <attribute name="type" type="{http://www.w3.org/2001/XMLSchema}string" />
    *       <attribute name="manual-start" type="{http://www.w3.org/2001/XMLSchema}boolean" />
    *       <attribute name="socket-buffer-size" type="{http://www.w3.org/2001/XMLSchema}string" />
    *       <attribute name="socket-read-timeout" type="{http://www.w3.org/2001/XMLSchema}string" />
@@ -2628,6 +2629,8 @@ public static class GatewaySender {
     protected String remoteDistributedSystemId;
     @XmlAttribute(name = "parallel")
     protected Boolean parallel;
+    @XmlAttribute(name = "type")
+    protected String type;
     @XmlAttribute(name = "manual-start")
     protected Boolean manualStart;
     @XmlAttribute(name = "socket-buffer-size")
@@ -2820,6 +2823,29 @@ public void setParallel(Boolean value) {
       parallel = value;
     }
 
+    /**
+     * Gets the value of the parallel property.
+     *
+     * possible object is
+     * {@link String }
+     *
+     */
+
+    public String getType() {
+      return type;
+    }
+
+    /**
+     * Sets the value of the type property.
+     *
+     * allowed object is
+     * {@link String }
+     *
+     */
+    public void setType(String value) {
+      this.type = value;
+    }
+
     /**
      * Gets the value of the manualStart property.
      *

geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySender.java

@@ -17,8 +17,6 @@
 import java.util.List;
 
 import org.apache.geode.annotations.Immutable;
-import org.apache.geode.internal.lang.SystemProperty;
-import org.apache.geode.internal.lang.SystemPropertyHelper;
 import org.apache.geode.util.internal.GeodeGlossary;
 
 /**
@@ -155,32 +153,6 @@ public interface GatewaySender {
   int CONNECTION_RETRY_INTERVAL = Integer
       .getInteger(GeodeGlossary.GEMFIRE_PREFIX + "gateway-connection-retry-interval", 1000);
 
-  /**
-   * Number of times to retry to get events for a transaction from the gateway sender queue when
-   * group-transaction-events is set to true.
-   * When group-transaction-events is set to true and a batch ready to be sent does not contain
-   * all the events for all the transactions to which the events belong, the gateway sender will try
-   * to get the missing events of the transactions from the queue to add them to the batch
-   * before sending it.
-   * If the missing events are not in the queue when the gateway sender tries to get them
-   * it will retry for a maximum of times equal to the value set in this parameter before
-   * delivering the batch without the missing events and logging an error.
-   * Setting this parameter to a very low value could cause that under heavy load and
-   * group-transaction-events set to true, batches are sent with incomplete transactions. Setting it
-   * to a high value could cause that under heavy load and group-transaction-events set to true,
-   * batches are held for some time before being sent.
-   */
-  int GET_TRANSACTION_EVENTS_FROM_QUEUE_RETRIES =
-      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "get-transaction-events-from-queue-retries",
-          10);
-  /**
-   * Milliseconds to wait before retrying to get events for a transaction from the
-   * gateway sender queue when group-transaction-events is true.
-   */
-  int GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS =
-      SystemProperty.getProductIntegerProperty(
-          SystemPropertyHelper.GET_TRANSACTION_EVENTS_FROM_QUEUE_WAIT_TIME_MS).orElse(1);
-
   /**
    * The order policy. This enum is applicable only when concurrency-level is > 1.
    *
@@ -418,10 +390,13 @@ enum OrderPolicy {
    */
   boolean isParallel();
 
+  String getType();
+
   /**
    * Returns groupTransactionEvents boolean property for this GatewaySender.
    *
    * @return groupTransactionEvents boolean property for this GatewaySender
+   * @deprecated use {@link #getType()}.
    *
    */
   boolean mustGroupTransactionEvents();

geode-core/src/main/java/org/apache/geode/cache/wan/GatewaySenderFactory.java

@@ -49,6 +49,8 @@ public interface GatewaySenderFactory {
    */
   GatewaySenderFactory setGroupTransactionEvents(boolean groupTransactionEvents);
 
+  GatewaySenderFactory setType(String type);
+
   /**
    * Adds a <code>GatewayEventFilter</code>
    *
@@ -188,7 +190,8 @@ public interface GatewaySenderFactory {
    *
    * @param filter The <code>GatewayEventSubstitutionFilter</code>
    */
-  GatewaySenderFactory setGatewayEventSubstitutionFilter(GatewayEventSubstitutionFilter filter);
+  GatewaySenderFactory setGatewayEventSubstitutionFilter(
+      GatewayEventSubstitutionFilter<?, ?> filter);
 
   /**
    * If true, receiver member id is checked by all dispatcher threads when the connection is