DistributedCacheOperation { public long startOperation() { DistributedRegion region = getRegion(); long viewVersion = -1; try { if (this.containsRegionContentChange()) { viewVersion = region.getDistributionAdvisor().startOperation(); } if (logger.isTraceEnabled(LogMarker.STATE_FLUSH_OP_VERBOSE)) { logger.trace(LogMarker.STATE_FLUSH_OP_VERBOSE, "dispatching operation in view version {}", viewVersion); } try { _distribute(); } catch (InvalidVersionException e) { if (logger.isTraceEnabled(LogMarker.DM_VERBOSE)) { logger.trace(LogMarker.DM_VERBOSE, "PutAll failed since versions were missing; retrying", e); } if (test_InvalidVersionAction != null) { test_InvalidVersionAction.run(); } _distribute(); } } catch (RuntimeException | Error e) { endOperation(viewVersion); throw e; } return viewVersion; } DistributedCacheOperation(CacheEvent event); static void writeValue(final byte deserializationPolicy, final Object vObj,
      final byte[] vBytes, final DataOutput out); static byte valueIsToDeserializationPolicy(boolean oldValueIsSerialized); static String deserializationPolicyToString(byte policy); boolean supportsDirectAck(); boolean supportsMulticast(); boolean canBeSentDuringShutdown(); boolean containsRegionContentChange(); long startOperation(); void endOperation(long viewVersion); void distribute(); @Override String toString(); static void setBeforePutOutgoing(Runnable beforePutOutgoing); @MutableForTesting
  static double LOSS_SIMULATION_RATIO; @MutableForTesting
  static Random LOSS_SIMULATION_GENERATOR; @MutableForTesting
  static long SLOW_DISTRIBUTION_MS; static final byte DESERIALIZATION_POLICY_NONE; static final byte DESERIALIZATION_POLICY_LAZY; static final byte VALUE_IS_BYTES; static final byte VALUE_IS_SERIALIZED_OBJECT; static final byte VALUE_IS_OBJECT; static final byte DESERIALIZATION_POLICY_NUMBITS; static final short DESERIALIZATION_POLICY_END; static final short DESERIALIZATION_POLICY_MASK; @MutableForTesting
  static boolean testSendingOldValues; @MutableForTesting
  static volatile DelayedAction test_InvalidVersionAction; }
@Test public void endOperationIsInvokedOnDistributionError() { DistributedRegion region = mock(DistributedRegion.class); CacheDistributionAdvisor advisor = mock(CacheDistributionAdvisor.class); when(region.getDistributionAdvisor()).thenReturn(advisor); TestOperation operation = new TestOperation(null); operation.region = region; try { operation.startOperation(); } catch (RuntimeException e) { assertEquals("boom", e.getMessage()); } assertTrue(operation.endOperationInvoked); }