CLOVER🍀

That was when it all began.

Infinispanのデータバージョニング

Infinispan

前に、Hot Rod ClientでVersionedAPIというものを使ったサンプルを書いたことがありますが、それとは違うバージョニングのためのAPIがあります。

Data Versioning
https://docs.jboss.org/author/display/ISPN/Data+Versioning

こちらは、Embedded Cacheで使用します。バージョン情報は、AdvancedCache#getCacheEntryから取得できる、CacheEntryから取得できるようです。

というわけで、ここからバージョン情報を取得することを目標に頑張ってみます。が、これが予想外に大変なことに…。

上記ページによると、データバージョニングの種類は

  • Simple
  • Partition-aware
  • External

の3種類があり、

  • Simpleは、long値のインクリメント（要は、バージョン管理番号）
  • Partition-awareは、ベクタクロック

を使用してバージョニングを行うようです。ExternalはInfinispan外のデータバージョニングのようなので、なんか毛色が違いそうです。

動作確認用のコードとして、以下のようなものを用意しました。
src/main/scala/InfinispanDataVersioning.scala

import scala.collection.JavaConverters._

import org.infinispan.manager.DefaultCacheManager
import org.infinispan.interceptors.VersionedEntryWrappingInterceptor

object InfinispanDataVersioning {
  def main(args: Array[String]): Unit = {
    val manager = new DefaultCacheManager("infinispan.xml")
    val cache = manager.getCache[String, String]()
    val advancedCache = cache.getAdvancedCache

    for (
      i <- advancedCache.getInterceptorChain.asScala
      if i.isInstanceOf[VersionedEntryWrappingInterceptor]
    ) println(s"Versioned Interceptor => $i")

    try {
      cache.put("key1", "value1")
      println(s"key1 => ${cache.get("key1")}")
      cache.put("key1", "value2")
      println(s"key1 => ${cache.get("key1")}")
      cache.put("key1", "value3")
      println(s"key1 => ${cache.get("key1")}")

      println("-----------------------------------------")

      val cacheEntry = advancedCache.getCacheEntry("key1", null, null)
      println(s"entry => $cacheEntry")
      println(s"entry key: ${cacheEntry.getKey}, entry value: ${cacheEntry.getValue}")
      val versionEntry = cacheEntry.getVersion
      println(versionEntry)
    } finally {
      cache.stop()
      manager.stop()
    }
  }
}

いろいろ確認していて分かったのですが、Data Versioningが有効になった時には、CacheのインターセプターにVersionedEntryWrappingInterceptorが追加されることがわかったので、それを確認しています。

    for (
      i <- advancedCache.getInterceptorChain.asScala
      if i.isInstanceOf[VersionedEntryWrappingInterceptor]
    ) println(s"Versioned Interceptor => $i")

また、先に書いた通り、バージョンに関する情報はCacheEntryから取得できるので、こちらも出力するようにしています。

      val cacheEntry = advancedCache.getCacheEntry("key1", null, null)
      println(s"entry => $cacheEntry")
      println(s"entry key: ${cacheEntry.getKey}, entry value: ${cacheEntry.getValue}")
      val versionEntry = cacheEntry.getVersion
      println(versionEntry)

Infinispanの設定ファイルは、こちら。
src/main/resources/infinispan.xml

<?xml version="1.0" encoding="UTF-8"?>
<infinispan
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="urn:infinispan:config:5.2 http://www.infinispan.org/schemas/infinispan-config-5.2.xsd"
    xmlns="urn:infinispan:config:5.2">

  <global>
    <transport clusterName="infinispan-data-versioning">
      <properties>
        <property name="configurationFile" value="jgroups.xml" />
      </properties>
    </transport>
    <globalJmxStatistics
        enabled="true"
        jmxDomain="org.infinispan"
        cacheManagerName="DefaultCacheManager"
        />
  </global>

  <default>
    <jmxStatistics enabled="true"/>
    <clustering mode="distribution">
      <hash numOwners="2" />
      <sync />
    </clustering>

    <versioning
        enabled="false"
        type="SIMPLE"
        useTombstones="false"
        tombstoneLifespan="60000" />
  </default>
</infinispan>

とりあえず、何も考えずにSimpleを使うように設定しています。
＊JGroupsの設定は、端折ります

では、実行。

> run-main InfinispanDataVersioning
[info] Running InfinispanDataVersioning 
[error] org.infinispan.config.ConfigurationException: javax.xml.stream.XMLStreamException: ParseError at [row,col]:[31,37]
[error] Message: Unexpected attribute 'type' encountered
[error] 	at org.infinispan.configuration.parsing.ParserRegistry.parse(ParserRegistry.java:87)
[error] 	at org.infinispan.manager.DefaultCacheManager.<init>(DefaultCacheManager.java:353)
[error] 	at org.infinispan.manager.DefaultCacheManager.<init>(DefaultCacheManager.java:326)
  〜省略〜
[error] 	at scala.tools.nsc.MainGenericRunner.main(MainGenericRunner.scala)
[error] Caused by: javax.xml.stream.XMLStreamException: ParseError at [row,col]:[31,37]
[error] Message: Unexpected attribute 'type' encountered
[error] 	at org.infinispan.configuration.parsing.ParseUtils.unexpectedAttribute(ParseUtils.java:76)
  〜省略〜
[error] 	at org.infinispan.configuration.parsing.Parser52.readElement(Parser52.java:98)
[error] 	at org.infinispan.configuration.parsing.Parser52.readElement(Parser52.java:75)
[error] 	at org.jboss.staxmapper.XMLMapperImpl.processNested(XMLMapperImpl.java:110)
[error] 	at org.jboss.staxmapper.XMLMapperImpl.parseDocument(XMLMapperImpl.java:69)
[error] 	at org.infinispan.configuration.parsing.ParserRegistry.parse(ParserRegistry.java:77)
[error] 	... 22 more
java.lang.RuntimeException: Nonzero exit code returned from runner: 1
	at scala.sys.package$.error(package.scala:27)
[trace] Stack trace suppressed: run last compile:run-main for the full output.
[error] (compile:run-main) Nonzero exit code returned from runner: 1
[error] Total time: 1 s, completed 2013/04/06 23:04:06

なんか、盛大にエラーを吐いてコケてくれました。

というか、

[error] Message: Unexpected attribute 'type' encountered

「type属性なんて知らないんだけど」と、明らかにおかしいことを言われています。

ここで、XML Schemaの定義を見てみます。
http://docs.jboss.org/infinispan/5.2/configdocs/urn_infinispan_config_5.2/complexType/configuration.versioning.html

Model
    <versioning
      enabled = xs:boolean : false
      versioningScheme = tns:versioningScheme>
    </versioning>

…なんか、全然違う属性になってますが。

ところで、Infinispanには直接関係ありませんけど、XML Schemaの内容がこういうAPI風の形で確認できるって便利ですね。今回は、これに随分助けられました。

で、対応するVersioningSchemeを見てみると
VersioningScheme
http://docs.jboss.org/infinispan/5.2/apidocs/org/infinispan/configuration/cache/VersioningScheme.html
「NONE」と「SIMPLE」しか指定できそうにないんですけどー。

Infinispan 5.1からの機能なのに、5.2で再構成したんでしょうか？少なくとも、ドキュメントは古い、で間違いなさそうです。

こうなると、「SIMPLE」しか使えないことは明らかなので、これを動かすことを目標にして頑張ってみました。

最終的に確認の助けになったのは、Infinispanのテストコードですね。
https://github.com/infinispan/infinispan/blob/master/core/src/test/java/org/infinispan/statetransfer/BaseOperationsDuringStateTransferTest.java

   @Override
   protected void createCacheManagers() {
      cacheConfigBuilder = getDefaultClusteredCacheConfig(cacheMode, isTransactional, true);
      if (isTransactional) {
         cacheConfigBuilder.transaction().transactionMode(TransactionMode.TRANSACTIONAL)
               .transactionManagerLookup(new DummyTransactionManagerLookup())
               .syncCommitPhase(true).syncRollbackPhase(true);

         if (isOptimistic) {
            cacheConfigBuilder.transaction().lockingMode(LockingMode.OPTIMISTIC)
                  .locking().writeSkewCheck(true).isolationLevel(IsolationLevel.REPEATABLE_READ)
                  .versioning().enable().scheme(VersioningScheme.SIMPLE);
         } else {
            cacheConfigBuilder.transaction().lockingMode(LockingMode.PESSIMISTIC);
         }
      }

んー、これを参考にXML設定を書くと、こんな感じかな。

<?xml version="1.0" encoding="UTF-8"?>
<infinispan
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="urn:infinispan:config:5.2 http://www.infinispan.org/schemas/infinispan-config-5.2.xsd"
    xmlns="urn:infinispan:config:5.2">

  <global>
    <transport clusterName="infinispan-data-versioning">
      <properties>
        <property name="configurationFile" value="jgroups.xml" />
      </properties>
    </transport>
    <globalJmxStatistics
        enabled="true"
        jmxDomain="org.infinispan"
        cacheManagerName="DefaultCacheManager"
        />
  </global>

  <default>
    <jmxStatistics enabled="true"/>
    <clustering mode="distribution">
      <hash numOwners="2" />
      <sync />
    </clustering>

    <versioning
        enabled="true"
        versioningScheme="SIMPLE" />

    <transaction
        transactionManagerLookupClass="org.infinispan.transaction.lookup.GenericTransactionManagerLookup"
        transactionMode="TRANSACTIONAL"
        lockingMode="OPTIMISTIC"
        autoCommit="true" />
    <locking
        isolationLevel="REPEATABLE_READ"
        writeSkewCheck="true" />
  </default>
</infinispan>

で、動かすと

> run-main InfinispanDataVersioning
  〜省略〜
[info] Versioned Interceptor => org.infinispan.interceptors.VersionedEntryWrappingInterceptor@3c4d63d0
[info] key1 => value1
[info] key1 => value2
[info] key1 => value3
[info] -----------------------------------------
[info] entry => ClusteredRepeatableReadEntry(b3fbd77){key=key1, value=value3, oldValue=null, isCreated=false, isChanged=false, isRemoved=false, isValid=true}
[info] entry key: key1, entry value: value3
[info] SimpleClusteredVersion{topologyId=0, version=3}
  〜省略〜
[success] Total time: 6 s, completed 2013/04/06 23:14:15

バージョン情報も取得できています。3回更新したので、バージョンは3ってことですね。

これ、ホントに最低限の設定で、例えば

    <locking
        isolationLevel="REPEATABLE_READ"
        writeSkewCheck="true" />

をコメントアウトしたりすると、

[info] key1 => value1
[info] key1 => value2
[info] key1 => value3
[info] -----------------------------------------
[info] entry => ImmortalCacheEntry{key=key1, value=ImmortalCacheValue {value=value3}}
[info] entry key: key1, entry value: value3
[info] null

インターセプターがいなくなり、バージョン情報も取得できなくなります。

ここで、改めて
https://docs.jboss.org/author/display/ISPN/Data+Versioning
に載っているSimple versioningの説明を読んで、設定すべき内容を確認すると…

  • クラスタリングモードでInfinispanを設定すること（distributionかreplicationは問わない）（＊ちょっと追記しました）
  • トランザクションを有効にすること（transactionMode="TRANSACTIONAL"）
  • ロックモードは楽観的ロックであること（lockingMode="OPTIMISTIC"）
  • ロックの設定を行い、トランザクション分離レベルを「REPEATABLE READ」にすること（isolationLevel="REPEATABLE_READ"）
  • write skewチェックを行うようにすること（writeSkewCheck="true"）
  • バージョニングを指定すること（versioningScheme="SIMPLE"）

ここまで設定して、はじめて「SIMPLE」のデータバージョニングが使用できるようになります。

（追記）
ちなみに、クラスタリングモードにせずに、ローカルのみでInfinispanを動かす時には、オブジェクトの参照でやっているみたいです。

When operating in LOCAL mode, write skew checks rely on Java object references to compare differences and this is adequate to provide a reliable write-skew check, however this technique is useless in a cluster and a more reliable form of versioning is necessary to provide reliable write skew checks.

https://docs.jboss.org/author/display/ISPN/Data+Versioning

なので、一応ローカルモードでもwrite-skewによるバージョンチェックはできないことはない、って感じでしょうか。まあ、本気のチェックではありませんが…。

とりあえず、今のInfinispanでは単純なlongのインクリメントによるバージョン管理が使用できるのみのようです。というか、これを有効にしないと楽観的ロックって実は機能しないような気がするのですが…気のせい？

Partition-awareは、そのうちまた登場するのかな？