Este artigo apresenta as operações básicas do cliente Java Curator para Zookeeper e como utilizá-lo para resolver problemas reais.

Acesso ao Zookeeper via Java

Para interagir com o Zookeeper em Java, os clientes mais comuns são o zkcleint e o Curator. O Curator oferece um nível de abstração mais elevado, simplificando significativamente o desenvolvimento. Isso levou à sua ampla adoção.

• Encapsula o tratamento da conexão entre o cliente Zookeeper e o servidor Zookeeper.
• Fornece uma API com estilo fluente para as operações.
• Oferece abstrações para diversos cenários de aplicação do Zookeeper (como bloqueios distribuídos e eleição de líder).

Dependência Maven

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-framework</artifactId>
    <version>5.2.0</version>
</dependency>

Estabelecendo a Conexão

O Curator permite a criação do cliente de duas formas: com estilo fluente ou com chamadas de métodos tradicionais.

public class CuratorApp {
    public static void main(String[] args) throws Exception {
        // Criação do cliente
        CuratorFramework client =
                CuratorFrameworkFactory.newClient("host-zk:2181", 5000, 20000,
                        new ExponentialBackoffRetry(1000, 3));
        client.start();
        client.blockUntilConnected();
        System.out.println("Conexão com Zookeeper estabelecida com sucesso.");

        // Leitura de dados de um nó
        byte[] nodeData = client.getData().forPath("/exemplo-no");
        System.out.println("Conteúdo do nó: " + new String(nodeData));
    }
}

Estratégias de Retry

O Curator implementa várias estratégias internas de retry:

• ExponentialBackoffRetry: Tenta novamente um número especificado de vezes, com intervalos crescentes entre tentativas. O intervalo é calculado por: baseSleepTimeMs * Math.max(1, random.nextInt(1 << (retryCount + 1))).
• RetryNTimes: Estratégia que limita o número máximo de tentativas.
• RetryOneTime: Realiza apenas uma tentativa adicional.
• RetryUntilElapsed: Continua tentando até que um tempo total decorrrido seja atingido.

Espaços de Nomes (Namespace)

Uma característica importante é o suporte a namespace, que isola a operação do cliente. Todas as operações nos nós do Zookeeper são relativas ao caminho definido como namespace, o que ajuda na separação lógica de diferentes aplicações ou serviços.

public class CuratorApp {
    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.builder()
                .connectString("host-zk:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(20000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace("meu-projeto") // Define o namespace
                .build();
        client.start();

        // Operações dentro do namespace "/meu-projeto"
        byte[] data = client.getData().forPath("/config");
        System.out.println("Dados da configuração: " + new String(data));
    }
}

Operações CRUD nos Nós

O código a seguir demonstra como realizar operações básicas de criação, leitura, atualização e exclusão de nós usando o Curator.

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

public class CuratorCrudDemo {
    private final CuratorFramework zkClient;

    public CuratorCrudDemo() {
        zkClient = CuratorFrameworkFactory.builder()
                .connectString("host-zk:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(20000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace("operacoes")
                .build();
        zkClient.start();
    }

    public void performCRUD() throws Exception {
        // 1. Criar um nó persistente
        String createdPath = zkClient.create()
                .creatingParentsIfNeeded()
                .withMode(CreateMode.PERSISTENT)
                .forPath("/meu-no");
        System.out.println("Nó criado em: " + createdPath);

        // 2. Ler dados e estatísticas do nó
        Stat nodeStat = new Stat();
        byte[] readData = zkClient.getData()
                .storingStatIn(nodeStat)
                .forPath(createdPath);
        System.out.println("Dados lidos: " + new String(readData));
        System.out.println("Versão do nó: " + nodeStat.getVersion());

        // 3. Atualizar dados do nó (casando a versão)
        Stat updatedStat = zkClient.setData()
                .withVersion(nodeStat.getVersion())
                .forPath(createdPath, "Novo conteúdo".getBytes());

        // 4. Ler dados atualizados
        String finalData = new String(zkClient.getData().forPath(createdPath));
        System.out.println("Dados após atualização: " + finalData);

        // 5. Excluir o nó
        zkClient.delete().forPath(createdPath);

        // 6. Verificar se o nó ainda existe
        Stat check = zkClient.checkExists().forPath(createdPath);
        if (check == null) {
            System.out.println("Nó excluído com sucesso.");
        }
    }

    public static void main(String[] args) throws Exception {
        CuratorCrudDemo demo = new CuratorCrudDemo();
        demo.performCRUD();
    }
}

Requisições Assíncronas

Em operações assíncronas, o cliente inicia a solicitação e uma thread separada a executa. O resultado é notificado através de um callback.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public void asyncOperationExample() throws Exception {
    CountDownLatch completionLatch = new CountDownLatch(2);
    ExecutorService callbackExecutor = Executors.newFixedThreadPool(2);

    // Criação assíncrona de um nó
    zkClient.create()
            .withMode(CreateMode.EPHEMERAL)
            .inBackground((client, event) -> {
                System.out.println("Nó criado assincronamente: " + event.getPath());
                completionLatch.countDown();
            }, callbackExecutor)
            .forPath("/no-assincrono");

    // Exclusão assíncrona do mesmo nó
    zkClient.delete()
            .inBackground((client, event) -> {
                System.out.println("Nó excluído assincronamente: " + event.getPath());
                completionLatch.countDown();
            }, callbackExecutor)
            .forPath("/no-assincrono");

    // Aguardar conclusão das operações
    completionLatch.await();
    callbackExecutor.shutdown();
}

Controle de Acesso (ACL) no Zookeeper

O Zookeeper armazena dados críticos para o estado de sistemas distribuídos. Para proteger esses dados contra acessos ou modificações indevidas, ele fornece um mecanismo de Controle de Acesso (ACL).

Uma ACL é definida pelo formato: esquema:id:permissão.

• Esquema (Scheme): Define a estratégia de autorização.
• ID (Objeto de Autorização): O usuário ou entidade à qual a permissão é concedida.
• Permissão (Permission): Os direitos de acesso concedidos.

As ACLs são aplicadas por nó. Um nó filho não herda as permissões do seu pai. Um cliente pode não ter acesso a um nó, mas ter acesso a um subnó.

Esquemas de Permissão

O Zookeeper suporta os seguintes esquemas:

• world: Padrão. Concede permissão a qualquer pessoa (anyone).
• auth: Refere-se a qualquer usuário autenticado. No CLI, use addauth digest user:senha para adicionar o usuário.
• digest: Autenticação por usuário:senha. A senha é enviada em texto claro, mas armazenada no Zookeeper como um hash SHA-1.
• ip: Controle de acesso baseado no endereço IP ou subnet do cliente (ex: ip:192.168.1.0/24).

Tipos de Permissão

As permissões específicas que podem ser concedidas são:

• c (create): Permite criar nós filhos.
• d (delete): Permite excluir nós filhos.
• r (read): Permite ler os dados do nó e listar seus filhos.
• w (write): Permite escrever/atualizar os dados do nó.
• a (admin): Permite definir ACLs no nó.

Exemplos no CLI (zkCli.sh)

Vejamos como configurar ACLs usando a interface de linha de comando.

Esquema 'world'

# Criar um nó (ACL padrão: world:anyone:cdrwa)
[zk: host-zk:2181(CONNECTED) 0] create /no-world
# Verificar ACL
[zk: host-zk:2181(CONNECTED) 1] getAcl /no-world
'world,'anyone
: cdrwa

# Modificar ACL (remover permissões de escrita e admin)
[zk: host-zk:2181(CONNECTED) 2] setAcl /no-world world:anyone:cdra
# Verificar nova ACL
[zk: host-zk:2181(CONNECTED) 3] getAcl /no-world
'world,'anyone
: cdra

Esquema 'ip'

# Conectar ao servidor com IP específico
./zkCli.sh -server 192.168.1.100:2181

# Criar nó e definir ACL para dois IPs
[zk: 192.168.1.100:2181(CONNECTED) 0] create /no-ip
[zk: 192.168.1.100:2181(CONNECTED) 1] setAcl /no-ip ip:127.0.0.1:cdrwa,ip:192.168.1.100:cdrwa
[zk: 192.168.1.100:2181(CONNECTED) 2] getAcl /no-ip
'ip,'127.0.0.1
: cdrwa
'ip,'192.168.1.100
: cdrwa

Esquema 'auth' e 'digest'

# Adicionar um usuário autenticado à sessão atual
[zk: host-zk:2181(CONNECTED) 0] addauth digest meuuser:minhasenha

# Criar nó e conceder permissão a esse usuário específico
[zk: host-zk:2181(CONNECTED) 1] create /no-auth
[zk: host-zk:2181(CONNECTED) 2] setAcl /no-auth auth:meuuser:cdrwa
[zk: host-zk:2181(CONNECTED) 3] getAcl /no-auth
'digest,'meuuser:<hash-da-senha>
: cdrwa

# Opcionalmente, usar digest diretamente com a senha hashada
[zk: host-zk:2181(CONNECTED) 4] setAcl /no-auth digest:meuuser:<hash-da-senha>:cdrwa
</hash-da-senha></hash-da-senha>

Nota: O hash da senha pode ser gerado programaticamente, por exemplo, usando SHA-1 em "meuuser:minhasenha" e codificando o resultado em Base64.

Usando ACL com Curator

O exemplo abaixo mostra como definir uma ACL usando o Curator Framework. É crucial que o cliente esteja autenticado para acessar nós protegidos.

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Id;
import org.apache.zookeeper.server.auth.DigestAuthenticationProvider;
import java.util.Collections;
import java.util.List;

public class CuratorAclDemo {
    private final CuratorFramework client;

    public CuratorAclDemo() {
        // Inclui a autorização digest na construção do cliente
        client = CuratorFrameworkFactory.builder()
                .connectString("host-zk:2181")
                .sessionTimeoutMs(20000)
                .connectionTimeoutMs(20000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .authorization("digest", "meuuser:minhasenha".getBytes())
                .namespace("seguranca")
                .build();
        client.start();
    }

    public void createNodeWithAcl() throws Exception {
        // Preparar a lista de ACLs
        String userDigest = DigestAuthenticationProvider.generateDigest("meuuser:minhasenha");
        Id authId = new Id("digest", userDigest);
        ACL acl = new ACL(ZooDefs.Perms.ALL, authId); // Todas as permissões
        List<ACL> aclList = Collections.singletonList(acl);

        // Criar o nó com a ACL definida
        String securedNode = client.create()
                .creatingParentsIfNeeded()
                .withACL(aclList, false) // false para não propagar ACLs para parents
                .forPath("/dado-seguro", "Informação confidencial".getBytes());
        System.out.println("Nó seguro criado: " + securedNode);

        // Ler os dados (a operação deve ser bem-sucedida)
        String retrievedData = new String(client.getData().forPath(securedNode));
        System.out.println("Dados recuperados: " + retrievedData);
    }

    public static void main(String[] args) throws Exception {
        CuratorAclDemo aclDemo = new CuratorAclDemo();
        aclDemo.createNodeWithAcl();
    }
}

Mecanismo de Escuta de Eventos

A escuta de eventos permite que aplicações se inscrevam para mudanças em nós específicos, possibilitando a implementação de recrusos como locks distribuídos, centros de registro e configuração dinâmica.

O Curator disponibiliza a classe CuratorCache (a partir do Zookeeper 3.6) como uma forma robusta de gerenciar caches e listeners. Para usá-la, adicione a dependência:

<dependency>
    <groupId>org.apache.curator</groupId>
    <artifactId>curator-recipes</artifactId>
    <version>5.2.0</version>
</dependency>

Exemplo com CuratorCache

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.cache.CuratorCache;
import org.apache.curator.framework.cache.CuratorCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;

public class EventWatcherDemo {
    private final CuratorFramework client;

    public EventWatcherDemo() {
        client = CuratorFrameworkFactory.builder()
                .connectString("host-zk:2181")
                .sessionTimeoutMs(20000)
                .connectionTimeoutMs(20000)
                .retryPolicy(new ExponentialBackoffRetry(1000, 3))
                .namespace("eventos")
                .build();
        client.start();
    }

    public void monitorNode(String nodePath) throws Exception {
        // Cria um cache para observar o caminho específico
        CuratorCache cache = CuratorCache.build(client, nodePath, CuratorCache.Options.SINGLE_NODE_CACHE);

        // Configura o listener
        CuratorCacheListener listener = CuratorCacheListener.builder()
                .forCreates(nodeData -> System.out.println("NÓ CRIADO: " + nodeData.getPath()))
                .forChanges((oldData, newData) -> System.out.println("NÓ ALTERADO: " + newData.getPath()))
                .forDeletes(oldData -> System.out.println("NÓ EXCLUÍDO: " + oldData.getPath()))
                .forInitialized(() -> System.out.println("Cache inicializado para: " + nodePath))
                .build();

        // Adiciona o listener ao cache e inicia o monitoramento
        cache.listenable().addListener(listener);
        cache.start();

        // Realiza operações para gerar eventos
        client.create().forPath(nodePath, "valor-inicial".getBytes());
        client.setData().forPath(nodePath, "valor-alterado".getBytes());
        client.delete().forPath(nodePath);

        // Mantém a aplicação rodando para ver os eventos
        Thread.sleep(5000);
        cache.close();
    }

    public static void main(String[] args) throws Exception {
        EventWatcherDemo demo = new EventWatcherDemo();
        demo.monitorNode("/no-observado");
    }
}