java - Zookeeper Service Discovery 接続が強制的に閉じられたキュレーター

Question

Curator と Zookeeper を使用してサービス ディスカバリを試しています。

サービスを登録しようとすると、次のことが発生します。

2014-09-23 16:36:09,755 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxnFactory@197] - Accepted socket connection from /127.0.0.1:52026
2014-09-23 16:36:09,759 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:ZooKeeperServer@868] - Client attempting to establish new session at /127.0.0.1:52026
2014-09-23 16:36:09,790 [myid:] - INFO  [SyncThread:0:ZooKeeperServer@617] - Established session 0x148a17bad590007 with negotiated timeout 40000 for client /127.0.0.1:52026
2014-09-23 16:36:11,735 [myid:] - WARN  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@362] - Exception causing close of session 0x148a17bad590007 due to java.io.IOException: An existing con
nection was forcibly closed by the remote host
2014-09-23 16:36:11,736 [myid:] - INFO  [NIOServerCxn.Factory:0.0.0.0/0.0.0.0:2181:NIOServerCnxn@1007] - Closed socket connection for client /127.0.0.1:52026 which had sessionid 0x148a17bad590007
2014-09-23 16:36:32,001 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x148a17bad590006, timeout of 40000ms exceeded
2014-09-23 16:36:32,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x148a17bad590006
2014-09-23 16:36:52,000 [myid:] - INFO  [SessionTracker:ZooKeeperServer@347] - Expiring session 0x148a17bad590007, timeout of 40000ms exceeded
2014-09-23 16:36:52,001 [myid:] - INFO  [ProcessThread(sid:0 cport:-1)::PrepRequestProcessor@494] - Processed session termination for sessionid: 0x148a17bad590007

関連するコードは次のとおりです: ( getDiscovery メソッドは、上記のエラーの原因となるものです)

private ServiceDiscovery<InstanceDetails> getDiscovery() {
        return ServiceDiscoveryBuilder.builder(InstanceDetails.class)
            .basePath(Config.basePath)
            .client(curatorFramework)
            .serializer(jacksonInstanceSerializer)
            .build();
    }

私はこれを次のように使用しています：

public void advertiseAvailability() {
    try {
        ServiceDiscovery<InstanceDetails> discovery = getDiscovery();
        discovery.start();
        discovery.registerService(getInstance()); // getInstance returns a ServiceInstance<InstanceDetails>
        discovery.close();
        System.out.println("Advertised");
    } catch (Exception e) {
        throw Throwables.propagate(e);
    }
}

以下は構成です

public class Config{
    public static final String basePath = "/";
    public static final String serviceName = "myTestService1";
    public static final String address = "127.0.0.1";
    public static final int port = 2181;
}

以下は、必要なcuratorFrameworkの初期化です

    CuratorFramework curatorFramework;
    curatorFramework = CuratorFrameworkFactory.builder()
        .connectionTimeoutMs(1000)
        .retryPolicy(new RetryNTimes(10, 500))
        .connectString(Config.address+":"+Config.port)
        .build();
    curatorFramework.start();
    new EnsurePath(Config.basePath).ensure(curatorFramework.getZookeeperClient());

jacksonInstanceSerializer は次のとおりです。

    this.jacksonInstanceSerializer = instanceSerializerFactory.getInstanceSerializer(
        new TypeReference<ServiceInstance<InstanceDetails>>() {}
    );

ここで、instanceSerializerFactory は次のとおりです。

public class InstanceSerializerFactory {
    private final ObjectReader objectReader;
    private final ObjectWriter objectWriter;

    public InstanceSerializerFactory(ObjectReader objectReader, ObjectWriter objectWriter) {
        this.objectReader = objectReader;
        this.objectWriter = objectWriter;
    }

    public <T> InstanceSerializer<T> getInstanceSerializer(
            TypeReference<ServiceInstance<T>> typeReference) {
        return new JacksonInstanceSerializer<T>(objectReader, objectWriter, typeReference);
    }

}

final class JacksonInstanceSerializer<T> implements InstanceSerializer<T> {
    private final TypeReference<ServiceInstance<T>> typeRef;
    private final ObjectWriter objectWriter;
    private final ObjectReader objectReader;

    JacksonInstanceSerializer(ObjectReader objectReader, ObjectWriter objectWriter,
                              TypeReference<ServiceInstance<T>> typeRef) {
        this.objectReader = objectReader;
        this.objectWriter = objectWriter;
        this.typeRef = typeRef;
    }

注: インスタンスの詳細は、登録しようとしている内容です (サービスの名前とアドレスなど)