I've implemented a version of Charlie's idea in the past. The data region
defines a TransactionListener that:
- creates a UnitOfWork object containing a List of Event objects. Each
Event contains:
- regionName
- operation
- key
- value
-puts into a transaction region that has a gateway sender attached to it.
It also has a CacheWriter attached to it.
On the remote site, the CacheWriter:
- begins a transaction
- iterates the Events and executes each one (different behavior depending
on the operation)
- commits the transaction
There are definitely some caveats:
- There is a race condition between the commit in the data region and the
TransactionListener afterCommit invocation doing the put into the
transaction region. If the server crashes after the put into the data
region but before the afterCommit callback, there will be data loss. In
that case, the transaction in question will not have been stored in the
transaction region and not be sent to the remote site.
- What do you do if a transaction fails in the remote site? I think that
means a similar remote transaction will fail in the local site. How do you
resolve this issue? You might want to look at GatewayConflictResolver
instead of this idea.
- The transaction region has to be cleared periodically. You can use
expiration here.
- As you noticed, knowing not to execute the transaction in the CacheWriter
in the local site is a bit tricky. isOriginRemote might not be the best
method of determining remote events. For example, if the transaction is
from a client, isOriginRemote will be be true even on the local site. I've
used the distributed-system-id as a callback argument in the put into the
transaction region. You might also be able to use groups. In the
CacheWriter, compare the local distributed system id to the callback
argument. If they are equal, skip the transaction processing.
TransactionListener.afterCommit:
public void afterCommit(TransactionEvent event) {
UnitOfWork uow = new UnitOfWork();
for (CacheEvent cacheEvent : event.getEvents()) {
EntryEvent entryEvent = (EntryEvent) cacheEvent;
uow.addEvent(cacheEvent.getRegion().getFullPath() ,
cacheEvent.getOperation(), entryEvent.getKey(), entryEvent.getNewValue());
}
this.cache.getRegion("transaction").put(event.getTransactionId(), uow,
this.cache.getInternalDistributedSystem().getDistributionManager().getDistributedSystemId());
}
CacheWriter.beforeCreate/beforeUpdate:
private void process(EntryEvent<TransactionId, UnitOfWork> event) {
if
(!event.getCallbackArgument().equals(this.cache.getInternalDistributedSystem().getDistributionManager().getDistributedSystemId()))
{
CacheTransactionManager ctm = this.cache.getCacheTransactionManager();
ctm.begin();
for (Event e : event.getNewValue().getEvents()) {
Region region = this.cache.getRegion(e.getRegionPath());
if (e.getOperation().isUpdate()) {
region.put(e.getKey(), e.getValue());
} else if (e.getOperation().isCreate()) {
region.create(e.getKey(), e.getValue());
} else if (e.getOperation().isDestroy()) {
region.destroy(e.getKey());
}
}
ctm.commit();
}
}
Thanks,
Barry Oglesby
On Mon, Jan 20, 2020 at 11:52 AM John Blum <jb...@pivotal.io> wrote:
> From within...
>
> class MyTransactionListener extends TransactionListenerAdapter {
>
> public void commit(TransactionEvent event) {
>
> List<CacheEvent<?, ?>> cacheEvents = event.getEvents();
>
> // Null check shouldn't be necessary but...
> // I don't trust the Geode API to do the right thing and...
> // I have seen issues like this before, so better safe than sorry!
> if (events != null) {
> cacheEvents.stream()
> .filter(Objects::nonNull) // Also should not be necessary but...
> .filter(CacheEvent::isOriginRemote)
> .ifPresent(cacheEvent -> // now persist the event or do
> whatever you need to do);
> }
> }
>
> -j
>
>
> On Mon, Jan 20, 2020 at 5:29 AM Alberto Bustamante Reyes
> <alberto.bustamante.re...@est.tech> wrote:
>
>> Thanks Charlie! In that case, what could be a good approach to
>> differentiate locally generated events from replicated events?
>> If you use a TransactionListener and a region to store and replicate
>> transaction objects, you would need that differentiation for not applying
>> twice the local transactions.
>> ------------------------------
>> *De:* Charlie Black <cbl...@pivotal.io>
>> *Enviado:* sábado, 18 de enero de 2020 0:32
>> *Para:* user@geode.apache.org <user@geode.apache.org>
>> *Asunto:* Re: WAN replication of transactions
>>
>> Take a look at Transaction Listener - the tranaction listener can get all
>> of the objects in the transaction. That could then be put into a
>> transaction object that gets transmitted over wan as single message. Just
>> do a put of all that you need to another region.
>>
>> Just a thought.
>>
>>
>>
>> On Fri, Jan 17, 2020 at 1:36 PM Alberto Bustamante Reyes
>> <alberto.bustamante.re...@est.tech> wrote:
>>
>> Thanks for your answers, thats what I understood from documentation.
>>
>> Then I see a potential point of failure. If a transaction in site-1 is
>> not completely sent to site-2, you have to ensure that applications using
>> the inconsistent data of site-2 are able to detect that or are prepared to
>> handle it.
>>
>> After reading the documentation, I was wondering if it could be possible
>> to send all events of a transaction using the same dispatcher thread of a
>> serial sender with "partition" policy taking into account that regions
>> involved on the transaction are colocated. Doing this, each batch of events
>> sent would contain almost only complete transactions.
>>
>> BR/
>>
>> Alberto
>> ------------------------------
>> *De:* Eric Shu <e...@pivotal.io>
>> *Enviado:* viernes, 17 de enero de 2020 19:35
>> *Para:* user@geode.apache.org <user@geode.apache.org>
>> *Asunto:* Re: WAN replication of transactions
>>
>> The transaction is not supported across the wan sites.
>>
>> If inconsistent state occurred due to a network problem between the wan
>> sites, the inconsistency would exist during the network outage. However,
>> once the network problem is resolved, the wan gateway would resume sending
>> the events again. Two sides would be eventually consistent.
>>
>> The persistence and redundancy mentioned by Anil would be useful if the
>> senders are down (HA scenario).
>>
>> Regards,
>> Eric
>>
>> On Fri, Jan 17, 2020 at 6:48 AM Anilkumar Gingade <aging...@pivotal.io>
>> wrote:
>>
>> Currently in Geode events are not transactional (WAN events,
>> Client/server subscription events). To enable guaranteed event delivery It
>> provides capability like persistence and redundancy.
>>
>> -Anil.
>>
>> On Fri, Jan 17, 2020 at 3:01 AM Alberto Bustamante Reyes
>> <alberto.bustamante.re...@est.tech> wrote:
>>
>> Hi all,
>>
>> I have a question about transactions and WAN replication. If a
>> transaction generates an event for each operation, is it possible to ensure
>> that those operations are received together on the other site?
>>
>> There is a risk of having a site with an inconsistent state if there is a
>> network problem. Lets say you have two sites, and communication between
>> them is broken when not all events of a transaction on site 1 where sent to
>> site 2. How are you dealing with this risk? Have someone experimented this?
>>
>> Thanks,
>>
>> BR/
>>
>> Alberto
>>
>>
>> --
>> Charlie Black | cbl...@pivotal.io
>>
>
>
> --
> -John
> Spring Data Team
>
