Salesforce Bulk API Sink Connector for Confluent Platform¶

Kafka Connect Salesforce Bulk API Sink Connector は、Salesforce を Apache Kafka® と統合します。コネクターでは、Kafka のトピックにあるレコードを使用して Salesforce SObject 上で CRUD 操作（挿入、アップデート、削除）が実行され、それらのレコードが Salesforce に書き込まれます。これらのレコードを、Salesforce PushTopic Source Connector を使用して、Kafka のトピックに書き込むことができます。

たとえば、2 つの Salesforce.com 組織またはインスタンスとして、インスタンス A とインスタンス B があるとします。データエンジニアが Salesforce Lead オブジェクトを A から B に同期するとします。データエンジニアは、PushTopic Source Connector を、Salesforce Lead をインスタンス A から単一の Kafka のトピックにストリーミングするよう構成してデプロイできます。一方、シンクコネクターは、Lead をそのトピックからインスタンス B にストリーミングするように構成できます。構成に応じて、Lead SObject に対するすべての変更を組織間で同期できます。このコネクターは、スタンドアロンおよび分散 Connect ワーカーで使用できます。

注釈

コネクターの salesforce.object プロパティは、操作対象の SObject の名前を示します。シンクコネクターの着信メッセージの構造とフォーマットは、Salesforce PushTopic Source Connector for Confluent Platform の出力フォーマットと同じです。

構成プロパティ¶

シンクコネクターの構成プロパティの網羅的なリストについては、「 Salesforce Bulk API Sink Connector 構成プロパティ」を参照してください。

考慮事項¶

Salesforce Bulk API Sink Connector を使用する場合は、以下に従います。

想定外のエラー¶

コネクターで Salesforce SObject に対する操作を実行している場合に、想定外のエラーが報告されることがあります。以下に、エラーが発生する理由のいくつかをリストします。

重複レコードの挿入を試行している。重複の判断ルールは、Salesforce で構成可能です。
Id フィールドが一致しないため、存在しないレコードに対する削除、アップデート、またはアップサート試行している。
以前削除した Id フィールド値と一致する Id フィールド値のフィールドに対する操作を試行している。

ID フィールドのセマンティクス¶

Salesforce Bulk API Sink Connector で、Salesforce PushTopic Source Connector for Confluent Platform で生成された Kafka のトピックのレコードが消費される場合、SObject の本文にある他のフィールドと兄弟である Id フィールドが含まれます。その Id は、レコードのストリーミング元である Salesforce 組織内でのみ有効であることに注意してください。その Id フィールドを基にしたアップサート、削除、アップデート操作を、別の Salesforce 組織で使用して試行すると、失敗の原因となります。Id フィールドは、Salesforce の内部で管理されるので、挿入の場合 Id フィールドは常に無視されます。アップサート操作は、外部 ID 構成オプションの salesforce.use.custom.id.field=true および salesforce.custom.id.field.name=<externalIdField> を設定して使用する必要があります。

ご用心

Salesforce 組織にまたがるアップデートおよび削除操作の場合は、Salesforce で外部 ID を構成する必要があります。また、カスタム ID は、両方の組織で、常に外部 ID としてマークされる必要があります。

入力トピックレコードフォーマット¶

入力トピックレコードフォーマットは、Salesforce PushTopic Source Connector for Confluent Platform によって出力トピックに書き込まれるレコードフォーマットと同じであることが想定されます。Kafka キー値は必須ではありません。例については、「Salesforce PushTopic Source Connector for Confluent Platform 」を参照してください。

読み取り専用フィールド¶

Salesforce SObject のフィールドは、挿入、アップデート、またはアップサートの操作による書き込みができないことがあります。これは、Salesforce 内で、フィールドに creatable=false または updatable=false 属性が設定されているためです。それらの属性が設定されたフィールドへの書き込みを試行する場合、シンクコネクターでは、操作全体が失敗するのではなく、該当するフィールドが操作対象から除外されます。この動作は、構成できません。

イベントタイプ¶

Salesforce Bulk API Sink Connector の Kafka レコードフォーマットには、_EventType フィールドが含まれています。レコードが Salesforce PushTopic Source Connector for Confluent Platform によって作成された場合は、このフィールドには、レコードを生成した PushTopic イベントのタイプが含まれます。タイプは、created、updated、deleted です。レコードを処理するときに、シンクコネクターは（デフォルトで）、_EventType を、構成されている SObject の insert、update、または delete 操作にマップします。この動作は、override.event.type=true および salesforce.sink.object.operation=<sink operation> フィールドでオーバーライドできます。イベントタイプのオーバーライドによって、そのレコードの _EventType フィールドが無視され、すべてのレコードで、salesforce.sink.object.operation が実行されます。

API の制限¶

Salesforce Bulk API Sink Connector は、実行するバッチ数、バッチあたりのレコード数、バッチの長さによって制限されます。制限の詳細については、『Bulk API の制限』を参照してください。
Salesforce Bulk API では、External Id フィールドの upsert 操作のみがサポートされています。

クイックスタート¶

このクイックスタートでは、Salesforce PushTopic Source Connector を使用してデータを Kafka に書き込み、Salesforce Bulk API Sink Connector を使用して、データを Kafka から Salesforce にエクスポートします。

注釈

Kafka Connect を Confluent Cloud に接続する方法の例については、「分散クラスター」を参照してください。

Kafka のトピックへの書き込み¶

「例 : Salesforce PushTopic Source Connector の構成」の手順を完了して、Kafka にデータを書き込みます。以下の点に注意してください。

Lead を作成した後、トピックで Lead オブジェクトを確認する前に、Lead の名前を編集します。Kafka のトピックのコンテンツをチェックすると、2 件のレコード（_EventType created および updated の各レコード）が表示されます。
「Salesforce アカウント」の手順を実行して、シンクコネクターとは別の Salesforce 組織に 2 つ目のアカウントを作成します。このアカウントは、シンクコネクターの構成に使用されます。

コネクターの構成¶

前提条件

Confluent Platform
Confluent CLI （個別のインストールが必要）

注釈

Salesforce Bulk API Source Connector によって生成されたレコードを処理するために、以下の Single Message Transform（SMT）をコネクター構成に追加します。

"transforms" : "InsertField",
"transforms.InsertField.type" : "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.InsertField.static.field" : "_EventType",
"transforms.InsertField.static.value" : "created"

以下のコンテンツの salesforce-bulk-api-leads-sink-config.json という名前の構成ファイルを作成します。必ず実際のユーザー名、パスワード、セキュリティトークン、コンシューマ鍵、コンシューマの秘密を入力してください。この構成プロパティの詳細については、構成プロパティを参照してください。

{
   "name" : "SalesforceBulkApiSinkConnector",
   "config" : {

     "connector.class" : "io.confluent.connect.salesforce.SalesforceBulkApiSinkConnector",
     "tasks.max" : "1",
     "topics" : "sfdc-pushtopic-lead",
     "salesforce.object" : "Lead",
     "salesforce.password" : "< Required Configuration >",
     "salesforce.password.token" : "< Required Configuration >",
     "salesforce.username" : "< Required Configuration: secondary organization username >",
     "reporter.result.topic.replication.factor" : "1",
     "reporter.error.topic.replication.factor" : "1",
     "reporter.bootstrap.servers" : "localhost:9092",
     "confluent.topic.bootstrap.servers": "localhost:9092",
     "confluent.topic.replication.factor": "1",
     "confluent.license": " Omit to enable trial mode "
   }
 }

注釈

このコネクターを Kafka Connect Reporter と併せて使用する方法の詳細については、「Connect Reporter 」を参照してください。

Confluent CLI コマンドを入力して Salesforce Sink Connector を起動します。

ご用心

トピック名とフラグの間には、ダブルダッシュ（ -- ）を入れる必要があります。詳細については、こちらの投稿を参照してください。

ちなみに

Confluent CLI 開発用コマンドのコマンド構文は、5.3.0 で変更されています。該当するコマンドは confluent local に移行されています。たとえば、confluent start の構文は、confluent local services start に変わりました。詳細については、「confluent local 」を参照してください。

confluent local load SalesforceBulkApiSinkConnector -- -d salesforce-bulk-api-leads-sink-config.json

出力は以下のようになります。

{
   "name": "SalesforceBulkApiSinkConnector",
    "config": {
        "connector.class" : "io.confluent.connect.salesforce.SalesforceBulkApiSinkConnector",
        "tasks.max" : "1",
        "topics" : "sfdc-pushtopic-leads",
        "salesforce.object" : "Lead",
        "salesforce.username" : "<Required>"
        "salesforce.password" : "<Required>",
        "salesforce.password.token" : "<Required>",
        "reporter.result.topic.replication.factor" : "1",
        "reporter.error.topic.replication.factor" : "1",
        "reporter.bootstrap.servers" : "localhost:9092",
        "confluent.topic.bootstrap.servers": "localhost:9092",
        "confluent.topic.replication.factor": "1",
        "confluent.license": " Omit to enable trial mode "
    },
    "tasks": [
        ...
    ],
    "type": null
}

ちなみに

tasks フィールドには、開始済みのタスクについての情報が含まれます。

Salesforce での Lead の表示¶

セカンダリの Salesforce 組織にログインし、Lead オブジェクトが正しい名前で表示されることを検証します。これはプライマリの Salesforce 組織と一致します。

両方のコネクターの同時実行¶

シンクコネクターとソースコネクターの両方を別々のワーカーで実行すると、SObject の変更をほぼリアルタイムで同期できます。スタンドアロンモードで実行する場合は、いずれかのワーカーに rest.port 構成のプロパティを使用してカスタムポートを追加します。
プライマリ組織で必要に応じて Lead を追加および変更します。ソースコネクターによって変更がキャプチャーされ、トピックに書き込まれます。これは、シンクコネクターによって読み取られ、操作のソースとして使用されたトピックです。

その他のドキュメント¶

Salesforce Bulk API Sink Connector 構成プロパティ