第8章 パーティション処理の設定

スプリットブレインの状況では、各ネットワークパーティションは独自の JGroups ビューをインストールし、他のパーティションからノードを削除します。パーティションが互いに認識していないため、2 つ以上のパーティションに分割されているかどうかを直接判断する方法はありません。代わりに、明示的な脱退メッセージを送信せずに 1 つ以上のノードが JGroups クラスターから消えたときに、クラスターが分割されたと想定します。

前項で説明したように、Data Grid は、プロセス/マシンがクラッシュするため、またはネットワーク障害が原因でノードが JGroups ビューを残すかどうかを検出できません。ノードが JGroups クラスターを突然とすると、ネットワークの問題が原因でノードが JGroups クラスターを終了すると想定されます。

設定したコピー数 (numOwners) が 1 よりも大きい場合、クラスターは利用可能なままになり、クラッシュしたノードでデータの新しいレプリカを作成しようとします。ただし、リバランスプロセス中にその他のノードがクラッシュする可能性があります。短期間に numOwners ノードがクラッシュすると、一部のキャッシュエントリーがクラスターから完全に消える可能性があります。この場合、DENY_READ_WRITES または ALLOW_READS ストラテジーを有効にすると、(正しい)Data Grid はスプリットブレインのセクションに記載されているように、DEGRADED モードに入ります。

管理者は、numOwners を超えるノードを急速にシャットダウンして、それらのノードにのみ保存したデータが失われないようにすることもできます。管理者がノードを正常にシャットダウンすると、Data Grid はノードに戻ることができないことを認識します。ただし、クラスターは各ノードの残状況を追跡せず、それらのノードがクラッシュしたかのように、キャッシュが DEGRADED モードに入ります。

この段階では、クラスターが停止し、外部ソースからのデータを使用して再起動時にその状態を回復する方法はありません。クラスターは、numOwners の連続するノードの失敗を回避するために、適切な numOwners で設定されることが予想されます。したがって、この状況は非常に稀なはずです。アプリケーションがキャッシュ内の一部のデータを失うことが可能な場合、管理者は JMX 経由で可用性モードを AVAILABLE に戻すことができます。

競合マネージャーは、ユーザーが特定のキーに保存されているすべてのレプリカ値を取得できるツールです。保存したレプリカの値が競合しているキャッシュエントリーのストリームをユーザーが処理できるようにする他にもあります。さらに、EntryMergePolicy インターフェイスの実装を利用することで、そのような競合を自動的に解決できます。

パーティションが ConflictManager をマージすると、設定された EntryMergePolicy の競合を自動的に解決しようとしますが、アプリケーションが必要とする競合の有無を手動で検索または解決することもできます。

以下のコードは、EmbedCacheManager の ConflictManager を取得する方法、指定されたキーの全バージョンを取得する方法、および特定のキャッシュ全体で競合をチェックする方法を示しています。

EmbeddedCacheManager manager = new DefaultCacheManager("example-config.xml");
Cache<Integer, String> cache = manager.getCache("testCache");
ConflictManager<Integer, String> crm = ConflictManagerFactory.get(cache.getAdvancedCache());

// Get All Versions of Key
Map<Address, InternalCacheValue<String>> versions = crm.getAllVersions(1);

// Process conflicts stream and perform some operation on the cache
Stream<Map<Address, CacheEntry<Integer, String>>> conflicts = crm.getConflicts();
conflicts.forEach(map -> {
   CacheEntry<Integer, String> entry = map.values().iterator().next();
   Object conflictKey = entry.getKey();
   cache.remove(conflictKey);
});

// Detect and then resolve conflicts using the configured EntryMergePolicy
crm.resolveConflicts();

// Detect and then resolve conflicts using the passed EntryMergePolicy instance
crm.resolveConflicts((preferredEntry, otherEntries) -> preferredEntry);

キャッシュが分散またはレプリケートされない限り、パーティション処理の設定は無視されます。デフォルトのパーティション処理ストラテジーは ALLOW_READ_WRITES で、デフォルトの EntryMergePolicy は MergePolicies::PREFERRED_ALWAYS です。

<distributed-cache name="the-default-cache">
   <partition-handling when-split="ALLOW_READ_WRITES" merge-policy="PREFERRED_NON_NULL"/>
</distributed-cache>

プログラムを使用しても同じことができます。

ConfigurationBuilder dcc = new ConfigurationBuilder();
dcc.clustering().partitionHandling()
                    .whenSplit(PartitionHandling.ALLOW_READ_WRITES)
                    .mergePolicy(MergePolicy.PREFERRED_ALWAYS);

<distributed-cache name="the-default-cache">
   <partition-handling when-split="ALLOW_READ_WRITES" merge-policy="org.example.CustomMergePolicy"/>
</distributed-cache>

ConfigurationBuilder dcc = new ConfigurationBuilder();
dcc.clustering().partitionHandling()
                    .whenSplit(PartitionHandling.ALLOW_READ_WRITES)
                    .mergePolicy(new CustomMergePolicy());

public class CustomMergePolicy implements EntryMergePolicy<String, String> {

   @Override
   public CacheEntry<String, String> merge(CacheEntry<String, String> preferredEntry, List<CacheEntry<String, String>> otherEntries) {
      // decide which entry should be used

      return the_solved_CacheEntry;
   }

# list all necessary implementations of EntryMergePolicy with the full qualified name
org.example.CustomMergePolicy

キャッシュの可用性モードは、JMX で Cache MBean の属性として公開されます。属性は書き込み可能で、管理者はキャッシュを DEGRADED モードから AVAILABLE (一貫性のコスト) に強制的に移行することができます。

可用性モードは、AdvancedCache インターフェイスからもアクセスできます。

AdvancedCache ac = cache.getAdvancedCache();

// Read the availability
boolean available = ac.getAvailability() == AvailabilityMode.AVAILABLE;

// Change the availability
if (!available) {
   ac.setAvailability(AvailabilityMode.AVAILABLE);
}