Use --type=TxGroupingParallelGatewaySender... as suggested in jbarrett's review

Author: albertogpz

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,31 +153,7 @@ 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);
+  String DEFAULT_TYPE = "SerialGatewaySender";
 
   /**
    * The order policy. This enum is applicable only when concurrency-level is > 1.
@@ -418,10 +392,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>
    *

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

@@ -126,8 +126,6 @@ public abstract class AbstractGatewaySender implements InternalGatewaySender, Di
 
   protected boolean isParallel;
 
-  protected int retriesToGetTransactionEventsFromQueue;
-
   protected boolean isForInternalUse;
 
   protected boolean isDiskSynchronous;
@@ -255,7 +253,6 @@ public AbstractGatewaySender(InternalCache cache, StatisticsClock statisticsCloc
     alertThreshold = attrs.getAlertThreshold();
     copyDeprecatedAttributes(attrs);
     isParallel = attrs.isParallel();
-    retriesToGetTransactionEventsFromQueue = attrs.getRetriesToGetTransactionEventsFromQueue();
     isForInternalUse = attrs.isForInternalUse();
     diskStoreName = attrs.getDiskStoreName();
     remoteDSId = attrs.getRemoteDSId();
@@ -559,21 +556,12 @@ public boolean isParallel() {
     return isParallel;
   }
 
+  @Deprecated
   @Override
   public boolean mustGroupTransactionEvents() {
     return false;
   }
 
-  /**
-   * Returns retriesToGetTransactionEventsFromQueue int property for this GatewaySender.
-   *
-   * @return retriesToGetTransactionEventsFromQueue int property for this GatewaySender
-   *
-   */
-  public int getRetriesToGetTransactionEventsFromQueue() {
-    return retriesToGetTransactionEventsFromQueue;
-  }
-
   public boolean isForInternalUse() {
     return isForInternalUse;
   }
@@ -1008,11 +996,6 @@ private boolean checkForDistribution(EntryEventImpl event) {
     return true;
   }
 
-  public void distribute(EnumListenerEvent operation, EntryEventImpl event,
-      List<Integer> allRemoteDSIds) {
-    distribute(operation, event, allRemoteDSIds, false);
-  }
-
   public void distribute(EnumListenerEvent operation, EntryEventImpl event,
       List<Integer> allRemoteDSIds, boolean isLastEventInTransaction) {
 

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

@@ -66,8 +66,6 @@ public interface GatewaySenderAttributes {
 
   boolean mustGroupTransactionEvents();
 
-  int getRetriesToGetTransactionEventsFromQueue();
-
   boolean isForInternalUse();
 
   String getId();
@@ -91,4 +89,6 @@ public interface GatewaySenderAttributes {
   boolean isForwardExpirationDestroy();
 
   boolean getEnforceThreadsConnectSameReceiver();
+
+  String getType();
 }

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

@@ -81,8 +81,7 @@ public class GatewaySenderAttributesImpl implements MutableGatewaySenderAttribut
 
   private boolean groupTransactionEvents = GatewaySender.DEFAULT_MUST_GROUP_TRANSACTION_EVENTS;
 
-  private int retriesToGetTransactionEventsFromQueue =
-      GatewaySender.GET_TRANSACTION_EVENTS_FROM_QUEUE_RETRIES;
+  private String type = GatewaySender.DEFAULT_TYPE;
 
   private boolean isForInternalUse = GatewaySender.DEFAULT_IS_FOR_INTERNAL_USE;
 
@@ -178,8 +177,9 @@ public void setGroupTransactionEvents(boolean groupTransEvents) {
     groupTransactionEvents = groupTransEvents;
   }
 
-  public void setRetriesToGetTransactionEventsFromQueue(int retries) {
-    retriesToGetTransactionEventsFromQueue = retries;
+  public void setType(String type) {
+    this.type = type;
+    isParallel = type.equals("Parallel") ? true : false;
   }
 
   public void setForInternalUse(boolean forInternalUse) {
@@ -288,11 +288,6 @@ public boolean mustGroupTransactionEvents() {
     return groupTransactionEvents;
   }
 
-  @Override
-  public int getRetriesToGetTransactionEventsFromQueue() {
-    return retriesToGetTransactionEventsFromQueue;
-  }
-
   @Override
   public boolean isForInternalUse() {
     return isForInternalUse;
@@ -360,4 +355,9 @@ public boolean getEnforceThreadsConnectSameReceiver() {
     return enforceThreadsConnectSameReceiver;
   }
 
+  @Override
+  public String getType() {
+    return type;
+  }
+
 }

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

@@ -30,12 +30,4 @@ public interface InternalGatewaySenderFactory extends GatewaySenderFactory {
   void configureGatewaySender(GatewaySender senderCreation);
 
   GatewaySenderFactory setLocatorDiscoveryCallback(LocatorDiscoveryCallback myLocatorCallback);
-
-  /**
-   * Sets the maximum number of retries to get events from the queue
-   * to complete a transaction when groupTransactionEvents is true.
-   *
-   * @param retries the maximum number of retries.
-   */
-  GatewaySenderFactory setRetriesToGetTransactionEventsFromQueue(int retries);
 }

geode-core/src/main/java/org/apache/geode/internal/cache/xmlcache/ParallelAsyncEventQueueCreation.java

@@ -16,8 +16,6 @@
 
 import static org.apache.geode.internal.statistics.StatisticsClockFactory.disabledClock;
 
-import java.util.List;
-
 import org.apache.geode.CancelCriterion;
 import org.apache.geode.cache.wan.GatewaySender;
 import org.apache.geode.distributed.internal.DistributionAdvisee;
@@ -26,7 +24,6 @@
 import org.apache.geode.distributed.internal.DistributionManager;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
 import org.apache.geode.internal.cache.EntryEventImpl;
-import org.apache.geode.internal.cache.EnumListenerEvent;
 import org.apache.geode.internal.cache.InternalCache;
 import org.apache.geode.internal.cache.wan.AbstractGatewaySender;
 import org.apache.geode.internal.cache.wan.GatewaySenderAttributes;
@@ -38,10 +35,6 @@ public ParallelAsyncEventQueueCreation(InternalCache cache, GatewaySenderAttribu
     super(cache, disabledClock(), attrs);
   }
 
-  @Override
-  public void distribute(EnumListenerEvent operation, EntryEventImpl event,
-      List<Integer> remoteDSIds) {}
-
   @Override
   public void start() {}
 
@@ -56,6 +49,11 @@ public void rebalance() {
     throw new UnsupportedOperationException();
   }
 
+  @Override
+  public String getType() {
+    return "ParallelAsyncEventQueue";
+  }
+
   @Override
   public void fillInProfile(Profile profile) {}