ApsaraMQ for Kafka:Logstash からインターネット経由で ApsaraMQ for Kafka にメッセージを送信

SSL 証明書のダウンロード

Logstash の bin ディレクトリに切り替え、トラストストア証明書をダウンロードします：

cd <logstash-install-dir>/bin
wget -O kafka.client.truststore.jks https://github.com/AliwareMQ/aliware-kafka-demos/raw/master/kafka-log-stash-demo/vpc-ssl/mix.4096.client.truststore.jks

JAAS 設定ファイルの作成

Logstash の bin ディレクトリに jaas.conf という名前のファイルを作成し、次の内容を記述します：

KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="<your-username>"
  password="<your-password>";
};

プレースホルダーを実際の値に置き換えます：

Logstash 出力設定ファイルの作成

Logstash の bin ディレクトリに output.conf という名前のファイルを作成し、次の内容を記述します：

input {
    stdin {}
}

output {
    stdout { codec => json }
    kafka {
        bootstrap_servers => "<your-endpoint>"
        topic_id => "<your-topic>"
        security_protocol => "SASL_SSL"
        sasl_mechanism => "PLAIN"
        jaas_path => "<logstash-install-dir>/bin/jaas.conf"
        ssl_truststore_password => "KafkaOnsClient"
        ssl_truststore_location => "<logstash-install-dir>/bin/kafka.client.truststore.jks"
        ssl_endpoint_identification_algorithm => ""
    }
}

これらのプレースホルダーを実際の値に置き換えます：

残りのパラメーターは固定値を使用します。変更しないでください：

テストメッセージの送信

1. 出力設定を使用して Logstash を起動します：

      ./logstash -f output.conf

2. Logstash が起動したら、test と入力して Enter キーを押します。stdout 出力はメッセージをローカルに JSON 形式で表示し、kafka 出力はメッセージをご利用の ApsaraMQ for Kafka インスタンスに送信します。

   

パーティションステータスの確認

1. ApsaraMQ for Kafka コンソールにログインします。

2. [概要] ページの [リソース配布] セクションで、インスタンスが存在するリージョンを選択します。

3. [インスタンス] ページで、ターゲットインスタンスの名前をクリックします。

4. 左側のナビゲーションウィンドウで、[Topic] をクリックします。

5. Topic 名をクリックし、[Topic 詳細] ページで [パーティションステータス] タブをクリックします。

   パラメーター	説明
   パーティション ID	パーティションの ID。
   最小オフセット	パーティション内の最も古いオフセット。
   最大オフセット	パーティション内の最新のオフセット。
   メッセージ	パーティション内のメッセージの総数。
   最終更新日時	最新のメッセージが保存された時刻。

   

オフセットによるメッセージのクエリ

1. ApsaraMQ for Kafka コンソールにログインします。

2. [概要] ページの [リソース配布] セクションで、インスタンスが存在するリージョンを選択します。

3. [インスタンス] ページで、ターゲットインスタンスの名前をクリックします。

4. 左側のナビゲーションウィンドウで、[メッセージクエリ] をクリックします。

5. [検索方法] ドロップダウンリストから、[オフセットで検索] を選択します。

6. [Topic] ドロップダウンリストから Topic を選択し、[パーティション] ドロップダウンリストからパーティションを選択し、[オフセット] フィールドにオフセット値を入力して、[検索] をクリックします。コンソールは、指定された値以上のオフセットを持つすべてのメッセージを返します。たとえば、[パーティション] を 5 に、[オフセット] を 5 に設定すると、パーティション 5 からオフセット 5 以上のすべてのメッセージが返されます。

   パラメーター	説明
   パーティション	メッセージが保存されているパーティション。
   オフセット	パーティション内のメッセージオフセット。
   キー	メッセージキー。文字列として表示されます。
   [値]	メッセージ本文。文字列として表示されます。
   [作成日時]	メッセージが送信されたときのタイムスタンプ。ProducerRecord のタイムスタンプフィールドに値を指定した場合、その値が表示されます。値を指定しなかった場合は、メッセージが送信されたときのローカルシステム時刻が表示されます。タイムスタンプフィールドが 0 または無効な値に設定されている場合、時刻は 1970/x/x x:x:x 形式で表示されます。ApsaraMQ for Kafka バージョン 0.9 以前のクライアントでは、このフィールドを設定できません。
   [操作]	[キーのダウンロード]： メッセージキーをダウンロードします。[値のダウンロード]： メッセージ本文をダウンロードします。コンソールにはメッセージごとに最大 1 KB まで表示されます。1 KB を超える内容を表示するには、メッセージをダウンロードしてください。一度にダウンロードできるメッセージは最大 10 MB です。