Apache Pulsar に格納されているネームスペースやトピックのメタデータを、pulsar-flink-connector を用いて Java から取得する方法を紹介します。

Maven 依存関係

<dependency>
  <groupId>io.streamnative.connectors</groupId>
  <artifactId>pulsar-flink-connector-2.11-1.12</artifactId>
  <version>2.7.3</version>
</dependency>

<repositories>
  <repository>
    <id>central</id>
    <url>https://repo1.maven.org/maven2</url>
  </repository>
  <repository>
    <id>bintray-streamnative-maven</id>
    <url>https://dl.bintray.com/streamnative/maven</url>
  </repository>
</repositories>

コード例：PulsarMetadataReader の使用

以下は、PulsarMetadataReader を使って Pulsar のメタデータを読み取るサンプルです。

package com.example.pulsar;

import org.apache.flink.streaming.connectors.pulsar.internal.PulsarMetadataReader;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.auth.AuthenticationToken;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.apache.pulsar.common.schema.SchemaType;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;

public class PulsarCatalogReader {

    public static void main(String[] args) {
        ClientConfigurationData clientConfig = new ClientConfigurationData();
        clientConfig.setServiceUrl("pulsar://127.0.0.1:6650");
        clientConfig.setAuthentication(new AuthenticationToken("your-jwt-token-here"));

        try (PulsarMetadataReader metadataReader = new PulsarMetadataReader(
                "http://127.0.0.1:8443",
                clientConfig,
                "",
                new HashMap<>(),
                -1,
                -1)) {

            List<String> namespaces = metadataReader.listNamespaces();
            System.out.println("Namespaces: " + namespaces);

            for (String ns : namespaces) {
                List<String> topics = metadataReader.getTopics(ns);
                System.out.println("Topics in " + ns + ": " + topics);

                for (String topic : topics) {
                    SchemaInfo schema = metadataReader.getPulsarSchema(topic);
                    System.out.println("Schema Name: " + schema.getName());
                    System.out.println("Schema Type: " + schema.getType());
                    System.out.println("Schema Properties: " + schema.getProperties());
                    System.out.println("Schema Definition: " + schema.getSchemaDefinition());
                }
            }

        } catch (IOException | PulsarAdminException e) {
            e.printStackTrace();
        }
    }
}