[Oracle Cloud] Streaming で Key に関する挙動をみてみる

# はじめに

Oracle Cloud Infractructure(以下OCI)では、ストリーミングデータをリアルタイムに収集・処理が出来る Streaming というサービスが提供されています。Streaming が保持するデータは、実際のデータ(value) に対して、key を指定する形式で格納します。指定する key によって、Streaming の中のどこのパーティションに格納するかを、おおむねコントロールできます。

実際に、どのような挙動となるのか見ていきましょう。なお、Streaming には、Kafka 互換のAPIがあるので、Kafka Client 経由でデータを格納してみて、パーティションがどのように格納されるか確認します。



# 前提条件

`teststream1` という Stream は、次のように3つのパーティションで構成されています。

![1588703929401.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/104247/282bc775-34b0-d423-82c8-85cba2eebf87.png)



# Key Null

以下のJavaのコードで実行してみます。

```java
package jp.test.sugi;

import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

public class App {
    public static void main(final String[] args) {
        System.out.println("Start.");

        // 接続時の設定値を Properties インスタンスとして構築する
        final Properties properties = new Properties();

        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
                "cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092");
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        properties.put(SaslConfigs.SASL_JAAS_CONFIG,
                "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla\" password=\"8t[shwUN}I-d+{}8Nx_a\";");
        properties.put(CommonClientConfigs.RETRIES_CONFIG, 5);
        properties.put("max.request.size", 1024 * 1024);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // Producer を構築する
        final KafkaProducer<String, String> producer = new KafkaProducer<>(properties, new StringSerializer(),
                new StringSerializer());

        try {
            // トピックを指定してメッセージを送信する
            for (int i = 0; i < 10; i++) {
                String value = "message=" + i;
                ProducerRecord<String, String> record = new ProducerRecord<>("teststream01", value);

                Future<RecordMetadata> send = producer.send(record);
                RecordMetadata recordMetadata = send.get();

                System.out.print("partition: " + recordMetadata.partition() + ", ");
                System.out.print("topic: " + recordMetadata.topic() + ", ");
                System.out.println("value: " + value);
            }
        } catch (final Exception e) {
            System.out.println("例外が発生しました。");
            System.out.println(e);
        } finally {
            producer.close();
        }

        System.out.println("End.");
    }
}
```



実際のレコードを生成しているところはここです。Key を指定していないため Null になります。

```java
ProducerRecord<String, String> record = new ProducerRecord<>("teststream01", value);
```



実行結果を見ると、3つのパーティションに分散されているのが確認できました。

```
Start.
partition: 0, topic: teststream01, value: message=0
partition: 2, topic: teststream01, value: message=1
partition: 0, topic: teststream01, value: message=2
partition: 1, topic: teststream01, value: message=3
partition: 2, topic: teststream01, value: message=4
partition: 0, topic: teststream01, value: message=5
partition: 1, topic: teststream01, value: message=6
partition: 0, topic: teststream01, value: message=7
partition: 1, topic: teststream01, value: message=8
partition: 2, topic: teststream01, value: message=9
End.
```



# Key 指定

それでは次に Key を指定して、動きを確認します。Streaming では Kafka と同様にKey の ハッシュ値を基にして、格納するパーティションが決められます。同じ Key の時は、同じ Partition に配置されます。この挙動を確認するために、同じ key でデータを格納してみましょう。

code

```java
package jp.test.sugi;

import java.util.Properties;
import java.util.concurrent.Future;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.serialization.StringSerializer;

public class App {
    public static void main(final String[] args) {
        System.out.println("Start.");

        // 接続時の設定値を Properties インスタンスとして構築する
        final Properties properties = new Properties();

        properties.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG,
                "cell-1.streaming.ap-tokyo-1.oci.oraclecloud.com:9092");
        properties.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        properties.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        properties.put(SaslConfigs.SASL_JAAS_CONFIG,
                "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"poc02/oracleidentitycloudservice/suguru.sugiyama@oracle.com/ocid1.streampool.oc1.ap-tokyo-1.amaaaaaaycetm7yawtz56lnnerap4r45y4vheekgvhdaevxf3clfpuew6mla\" password=\"8t[shwUN}I-d+{}8Nx_a\";");
        properties.put(CommonClientConfigs.RETRIES_CONFIG, 5);
        properties.put("max.request.size", 1024 * 1024);
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);

        // Producer を構築する
        final KafkaProducer<String, String> producer = new KafkaProducer<>(properties, new StringSerializer(),
                new StringSerializer());

        try {
            // トピックを指定してメッセージを送信する
            for (int i = 0; i < 10; i++) {
                String value = "message=" + i;
                ProducerRecord<String, String> record = new ProducerRecord<>("teststream01", "key1", value);

                Future<RecordMetadata> send = producer.send(record);
                RecordMetadata recordMetadata = send.get();

                System.out.print("partition: " + recordMetadata.partition() + ", ");
                System.out.print("topic: " + recordMetadata.topic() + ", ");
                System.out.println("value: " + value);
            }
        } catch (final Exception e) {
            System.out.println("例外が発生しました。");
            System.out.println(e);
        } finally {
            producer.close();
        }

        System.out.println("End.");
    }
}
```



Key を指定しているのはここです。`key1` と指定しています。

```
ProducerRecord<String, String> record = new ProducerRecord<>("teststream01", "key1", value);
```



実行例を見てみると、全て Partition が 2 に偏っていることがわかります。

```
Start.
partition: 2, topic: teststream01, value: message=0
partition: 2, topic: teststream01, value: message=1
partition: 2, topic: teststream01, value: message=2
partition: 2, topic: teststream01, value: message=3
partition: 2, topic: teststream01, value: message=4
partition: 2, topic: teststream01, value: message=5
partition: 2, topic: teststream01, value: message=6
partition: 2, topic: teststream01, value: message=7
partition: 2, topic: teststream01, value: message=8
partition: 2, topic: teststream01, value: message=9
End.
```