This is an automated email from the ASF dual-hosted git repository.
zhouxj pushed a commit to branch feature/GEM-883
in repository https://gitbox.apache.org/repos/asf/geode.git
commit 5fa6ca55288c4efb83b07fdd77b80941bc74de47
Author: zhouxh <gz...@pivotal.io>
AuthorDate: Thu Nov 9 23:49:29 2017 -0800
GEODE-3967: if put hits concurrent modification exception should still
notify serial gateway sender
GEODE-3967: notifyTimestampsToGateways should inherit isConcurrencyConflict
GEODE-3967: add to secondary event isConcurrencyConflict
---
.../geode/internal/cache/AbstractRegionMap.java | 3 +++
.../geode/internal/cache/DestroyOperation.java | 3 ---
.../internal/cache/DistributedCacheOperation.java | 15 +++++++++++++-
.../apache/geode/internal/cache/LocalRegion.java | 19 +++++++++++++-----
.../geode/internal/cache/LocalRegionDataView.java | 9 +++++++++
.../wan/AbstractGatewaySenderEventProcessor.java | 15 +++++++++-----
.../internal/cache/wan/GatewaySenderEventImpl.java | 3 +++
.../serial/SerialGatewaySenderEventProcessor.java | 23 ++++++++++++++++++++++
8 files changed, 76 insertions(+), 14 deletions(-)
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java
index 404488b..75d8484 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/AbstractRegionMap.java
@@ -1183,6 +1183,7 @@ public abstract class AbstractRegionMap implements
RegionMap {
true/* conflict with clear */, duringRI, true);
doPart3 = true;
} catch (ConcurrentCacheModificationException ccme) {
+ event.isConcurrencyConflict(true);
VersionTag tag = event.getVersionTag();
if (tag != null && tag.isTimeStampUpdated()) {
// Notify gateways of new time-stamp.
@@ -2092,6 +2093,7 @@ public abstract class AbstractRegionMap implements
RegionMap {
}
} // !opCompleted
} catch (ConcurrentCacheModificationException ccme) {
+ event.isConcurrencyConflict(true);
VersionTag tag = event.getVersionTag();
if (tag != null && tag.isTimeStampUpdated()) {
// Notify gateways of new time-stamp.
@@ -2849,6 +2851,7 @@ public abstract class AbstractRegionMap implements
RegionMap {
clearOccured = true;
owner.recordEvent(event);
} catch (ConcurrentCacheModificationException ccme) {
+ event.isConcurrencyConflict(true);
VersionTag tag = event.getVersionTag();
if (tag != null && tag.isTimeStampUpdated()) {
// Notify gateways of new time-stamp.
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/DestroyOperation.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/DestroyOperation.java
index 0d2dc7f..7870f58 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/DestroyOperation.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/DestroyOperation.java
@@ -95,9 +95,6 @@ public class DestroyOperation extends
DistributedCacheOperation {
} catch (EntryNotFoundException e) {
dispatchElidedEvent(rgn, ev);
- if (!ev.isConcurrencyConflict()) {
- rgn.notifyGatewaySender(EnumListenerEvent.AFTER_DESTROY, ev);
- }
throw e;
} catch (CacheWriterException e) {
throw new Error(
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
index 908bf83..ddde23a 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/DistributedCacheOperation.java
@@ -1289,11 +1289,24 @@ public abstract class DistributedCacheOperation {
*/
protected void dispatchElidedEvent(LocalRegion rgn, EntryEventImpl ev) {
if (logger.isDebugEnabled()) {
- logger.debug("dispatching elided event: {}", ev);
+ logger.debug("GGG:dispatching elided event: {}", ev, new Exception());
}
ev.isConcurrencyConflict(true);
rgn.generateLocalFilterRouting(ev);
rgn.notifyBridgeClients(ev);
+ rgn.notifyGatewaySender(getOperation(ev), ev);
+ }
+
+ private EnumListenerEvent getOperation(EntryEventImpl ev) {
+ if (ev.getOperation().isInvalidate()) {
+ return EnumListenerEvent.AFTER_INVALIDATE;
+ } else if (ev.getOperation().isDestroy()) {
+ return EnumListenerEvent.AFTER_DESTROY;
+ } else if (ev.getOperation().isUpdate()) {
+ return EnumListenerEvent.AFTER_UPDATE;
+ } else {
+ return EnumListenerEvent.AFTER_CREATE;
+ }
}
protected abstract InternalCacheEvent createEvent(DistributedRegion rgn)
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
index aa0f8c6..38f74ec 100644
--- a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
+++ b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegion.java
@@ -2853,6 +2853,8 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
logger.debug("caught concurrent modification attempt when
applying {}", event);
}
notifyBridgeClients(event);
+ notifyGatewaySender(event.getOperation().isUpdate() ?
EnumListenerEvent.AFTER_UPDATE
+ : EnumListenerEvent.AFTER_CREATE, event);
}
if (!getDataView().isDeferredStats()) {
getCachePerfStats().endPut(startPut, event.isOriginRemote());
@@ -5630,6 +5632,9 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
logger.debug("caught concurrent modification attempt when applying
{}", event);
}
notifyBridgeClients(event);
+ notifyGatewaySender(event.getOperation().isUpdate() ?
EnumListenerEvent.AFTER_UPDATE
+ : EnumListenerEvent.AFTER_CREATE, event);
+
return false;
}
@@ -5856,6 +5861,9 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
updateTimeStampEvent.setGenerateCallbacks(false);
updateTimeStampEvent.distributedMember = event.getDistributedMember();
updateTimeStampEvent.setNewEventId(getSystem());
+ if (event.isConcurrencyConflict()) {
+ updateTimeStampEvent.isConcurrencyConflict(true);
+ }
if (event.getRegion() instanceof BucketRegion) {
BucketRegion bucketRegion = (BucketRegion) event.getRegion();
@@ -6117,8 +6125,7 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
}
protected void notifyGatewaySender(EnumListenerEvent operation,
EntryEventImpl event) {
- if (isPdxTypesRegion() || event.isConcurrencyConflict()) {
- // isConcurrencyConflict is usually a concurrent cache modification
problem
+ if (isPdxTypesRegion()) {
return;
}
@@ -6142,9 +6149,10 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
if (allRemoteDSIds != null) {
for (GatewaySender sender : getCache().getAllGatewaySenders()) {
if (allGatewaySenderIds.contains(sender.getId())) {
- // TODO: This is a BUG. Why return and not continue?
- if (!this.getDataPolicy().withStorage() && sender.isParallel()) {
- return;
+ // if isConcurrencyConflict is true, only notify serial gateway
sender
+ if ((!this.getDataPolicy().withStorage() ||
event.isConcurrencyConflict())
+ && sender.isParallel()) {
+ continue;
}
if (logger.isDebugEnabled()) {
logger.debug("Notifying the GatewaySender : {}", sender.getId());
@@ -6503,6 +6511,7 @@ public class LocalRegion extends AbstractRegion
implements LoaderHelperFactory,
if (event.getVersionTag() != null &&
!event.getVersionTag().isGatewayTag()) {
notifyBridgeClients(event);
}
+ notifyGatewaySender(EnumListenerEvent.AFTER_DESTROY, event);
return true; // event was elided
} catch (DiskAccessException dae) {
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
index eed6176..b68859e 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/LocalRegionDataView.java
@@ -25,6 +25,7 @@ import
org.apache.geode.internal.cache.entries.AbstractRegionEntry;
import org.apache.geode.internal.cache.tier.sockets.ClientProxyMembershipID;
import org.apache.geode.internal.cache.tier.sockets.VersionedObjectList;
import
org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
+import org.apache.geode.internal.logging.LogService;
/**
*
@@ -71,6 +72,10 @@ public class LocalRegionDataView implements InternalDataView
{
} catch (ConcurrentCacheModificationException e) {
// a newer event has already been applied to the cache. this can happen
// in a client cache if another thread is operating on the same key
+ event.isConcurrencyConflict(true);
+ LocalRegion lr = event.getLocalRegion();
+ LogService.getLogger().info("GGG:invalidateExistingEntry:" + event, new
Exception());
+ // lr.notifyGatewaySender(EnumListenerEvent.AFTER_INVALIDATE, event);
}
}
@@ -81,6 +86,10 @@ public class LocalRegionDataView implements InternalDataView
{
} catch (ConcurrentCacheModificationException e) {
// a later in time event has already been applied to the cache. this can
happen
// in a cache if another thread is operating on the same key
+ event.isConcurrencyConflict(true);
+ LocalRegion lr = event.getLocalRegion();
+ LogService.getLogger().info("GGG:updateEntryVersion:" + event, new
Exception());
+ // lr.notifyGatewaySender(EnumListenerEvent.TIMESTAMP_UPDATE, event);
}
}
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
index 7a2cee1..a557875 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/AbstractGatewaySenderEventProcessor.java
@@ -518,16 +518,21 @@ public abstract class AbstractGatewaySenderEventProcessor
extends Thread {
// version is < 7.0.1, especially to prevent another loop over
events.
if (!sendUpdateVersionEvents
&& event.getOperation() == Operation.UPDATE_VERSION_STAMP) {
- if (isTraceEnabled) {
- logger.trace(
- "Update Event Version event: {} removed from Gateway
Sender queue: {}", event,
- sender);
- }
+ logger.debug("Update Event Version event: {} removed from
Gateway Sender queue: {}",
+ event, sender);
itr.remove();
statistics.incEventsNotQueued();
continue;
}
+ if (((GatewaySenderEventImpl) event).isConcurrencyConflict) {
+ if (isDebugEnabled) {
+ logger.debug("primary should ignore the concurrency conflict
event:" + event);
+ }
+ itr.remove();
+ statistics.incEventsNotQueued();
+ continue;
+ }
boolean transmit = filter.beforeTransmit(event);
if (!transmit) {
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventImpl.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventImpl.java
index 5b1ba54..468dca2 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventImpl.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/GatewaySenderEventImpl.java
@@ -171,6 +171,8 @@ public class GatewaySenderEventImpl
protected boolean isInitialized;
+ public boolean isConcurrencyConflict = false;
+
/**
* Is this thread in the process of serializing this event?
*/
@@ -312,6 +314,7 @@ public class GatewaySenderEventImpl
if (initialize) {
initialize();
}
+ this.isConcurrencyConflict = event.isConcurrencyConflict();
}
/**
diff --git
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
index 734b560..995007d 100644
---
a/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
+++
b/geode-core/src/main/java/org/apache/geode/internal/cache/wan/serial/SerialGatewaySenderEventProcessor.java
@@ -291,6 +291,10 @@ public class SerialGatewaySenderEventProcessor extends
AbstractGatewaySenderEven
if (o != null && o instanceof GatewaySenderEventImpl) {
GatewaySenderEventImpl ge = (GatewaySenderEventImpl) o;
EventWrapper unprocessedEvent =
this.unprocessedEvents.remove(ge.getEventId());
+ if (unprocessedEvent != null && ge.isConcurrencyConflict) {
+ logger.info(
+ "GGG:secondary after removed by handleFailover:" +
unprocessedEvent + ":" + ge);
+ }
if (unprocessedEvent != null) {
unprocessedEvent.event.release();
if (this.unprocessedEvents.isEmpty()) {
@@ -379,6 +383,7 @@ public class SerialGatewaySenderEventProcessor extends
AbstractGatewaySenderEven
if (m != null) {
for (EventWrapper ew : m.values()) {
GatewaySenderEventImpl gatewayEvent = ew.event;
+ logger.info("GGG:releaseUnprocessedEvents:" + gatewayEvent);
gatewayEvent.release();
}
this.unprocessedEvents = null;
@@ -632,6 +637,10 @@ public class SerialGatewaySenderEventProcessor extends
AbstractGatewaySenderEven
return;
// now we can safely use the unprocessedEvents field
EventWrapper ew =
this.unprocessedEvents.remove(gatewayEvent.getEventId());
+ if (ew != null && gatewayEvent.isConcurrencyConflict) {
+ logger.info("GGG:secondary after removed by destroy listener:" + ew +
":" + gatewayEvent);
+ }
+
if (ew != null) {
ew.event.release();
statistics.incUnprocessedEventsRemovedByPrimary();
@@ -651,8 +660,16 @@ public class SerialGatewaySenderEventProcessor extends
AbstractGatewaySenderEven
return;
// now we can safely use the unprocessedEvents field
EventWrapper ew =
this.unprocessedEvents.remove(gatewayEvent.getEventId());
+ if (ew != null && gatewayEvent.isConcurrencyConflict) {
+ logger.info("GGG:secondary after removed by create listener:" + ew +
":" + gatewayEvent,
+ new Exception());
+ }
if (ew == null) {
+ if (gatewayEvent.isConcurrencyConflict) {
+ logger.info("GGG:secondary before add to by create listener:" +
gatewayEvent,
+ new Exception());
+ }
// first time for the event
if (logger.isTraceEnabled()) {
logger.trace("{}: fromPrimary event {} : {}->{} added to unprocessed
token map",
@@ -711,8 +728,14 @@ public class SerialGatewaySenderEventProcessor extends
AbstractGatewaySenderEven
// @todo add an assertion that !getPrimary()
// now we can safely use the unprocessedEvents field
Long v = this.unprocessedTokens.remove(gatewayEvent.getEventId());
+ if (v != null && gatewayEvent.isConcurrencyConflict) {
+ logger.info("GGG:secondary after removed token:" + v + ":" +
gatewayEvent);
+ }
if (v == null) {
+ if (gatewayEvent.isConcurrencyConflict) {
+ logger.info("GGG:secondary before add to:" + gatewayEvent, new
Exception());
+ }
// first time for the event
if (logger.isTraceEnabled()) {
logger.trace("{}: fromSecondary event {}:{}->{} added from
unprocessed events map",
--
To stop receiving notification emails like this one, please contact
"commits@geode.apache.org" <commits@geode.apache.org>.
