Provided some clarifications in the astra db sink doc

Author: ddieruf

modules/pulsar-io/pages/connectors/sinks/astra-db.adoc

@@ -5,9 +5,18 @@
 :page-tag: astra-db,cdc,sink-connector
 :page-aliases: docs@astra-streaming::connectors/sinks/astream-astradb-sink.adoc
 
-= Astra DB
+= Astra DB (Cassandra Enhanced)
 
-DataStax Astra DB Sink Connector is an enhanced version of the open-source Cassandra sink connector for Apache Pulsar.
+DataStax Astra DB Sink Connector is based on the open-source https://docs.datastax.com/en/pulsar-connector/docs/index.html[Cassandra sink connector for Apache Pulsar^]. Depending on how you deploy the connector, it can be used to sink topic messages with a table in Astra DB or a table in a Cassandra cluster outside of DB.
+
+The Astra Streaming portal provides simple way to connect this sink and a table in Astra DB with simply a token. Using pulsar-admin or the REST API, you can configure the sink to connect with a Cassandra connection manually.
+
+This reference assumes you are manually connecting to a Cassandra table.
+
+[TIP]
+====
+If you would like to see the code, refer to the https://github.com/datastax/pulsar-sink[open source here^]{external-link-icon}.
+====
 
 == Get Started
 
@@ -25,17 +34,19 @@ include::partial$connectors/sinks/monitoring.adoc[]
 
 There are two sets of parameters that support sink connectors.
 
-=== Astra Streaming
+=== Pulsar sink connector parameters
 
 [%header,format=csv,cols="2,1,1,3"]
 |===
 include::example$connectors/sinks/astra.csv[]
 |===
 
-=== Astra DB (configs)
+=== Cassandra Connection
 
 These values are provided in the *Configs* area.
 
+The "cloud.secureConnectBundle" can either be a path to your bundle zip or you can base64 encode the zip and provide it in the format: "base64:<b64 string>".
+
 [%header,format=csv,cols="2,1,1,3"]
 |===
 include::example$connectors/sinks/{connectorType}/config.csv[]
@@ -44,34 +55,65 @@ include::example$connectors/sinks/{connectorType}/config.csv[]
 // TODO: Need descriptions of every param
 === Auth Properties
 
+These values are provided in the *auth* area in the above cassandra connection parameters.
+
 [%header,format=csv,cols="2,1,1,3"]
 |===
 include::example$connectors/sinks/{connectorType}/auth.csv[]
 |===
 
-// TODO: This not really helpful. Need explanations, especially about mapping
 === Topic Properties
-[source,json,subs="attributes+"]
+
+These values are provided in the *topic* area in the above cassandra connection parameters.
+
+Refer to the official documentation for a https://docs.datastax.com/en/pulsar-connector/docs/cfgRefPulsarDseConnection.html[connection properties reference^].
+
+=== Mapping topic data to table columns
+
+[TIP]
+====
+There are quite a few examples in the "https://docs.datastax.com/en/pulsar-connector/docs/cfgPulsarMapTopicTable.html[Mapping pulsar topics to database tables^]" area of the official documentation
+====
+
+An essential part of using this sink connector is mapping message values to table columns.  There are many factors that influence how this done and what is possible. The 'mapping' string is a simple comma-separated list of column names and message value fields.
+
+While the getting started examples above show how to configure the connector in one large command, it is easier to manage this as a separate file. The following example show how to configure the connector using a configuration in json format. The "https://docs.datastax.com/en/pulsar-connector/docs/pulsarQuickStart.html[Pulsar Connector single instance quick start^]" guide provides a good example of this. Below are the minimum requirements.
+
+Create a file named configs.json using the following structure:
+
+[source]
 ----
-"<TABLE_NAME>": {
-    "<KEYSPACE>": {
-        "<TABLE_NAME>": {
-            "consistencyLevel": "LOCAL_QUORUM",
-            "deletesEnabled": true,
-            "mapping": "<Each field must be prefixed with either 'key' or 'value' or use the now() function. Example: 'part=value.name, id=value.id, num=value.number, fact=value.isfact, added=now()'>",
-            "nullToUnset": true,
-            "timestampTimeUnit": "MICROSECONDS",
-            "ttl": -1,
-            "ttlTimeUnit": "SECONDS"
+"archive": "builtin://cassandra-enhanced",
+"tenant": "<your tenant>",
+"namespace": "<your namespace>",
+"name": "<a really great name>",
+"inputs": ["<your topic name>"],
+"configs:": {
+  "topics": <the topic name (just the name, not the full address)>,
+  "cloud.secureConnectBundle": </path/to/secure-connect-database_name.zip OR base64 encode of the zip>,
+  "topic": {
+   "<the topic name (usually the same as above)>": {
+      "<the table keyspace name>": {
+        "<the table name>": {
+          "<all the connection properties>",
+          ...
+          "mapping": <field-to-column mapping>
         }
-    },
-    "codec": {
-        "date": "ISO_LOCAL_DATE",
-        "locale": "en_US",
-        "time": "ISO_LOCAL_TIME",
-        "timeZone": "UTC",
-        "timestamp": "CQL_TIMESTAMP",
-        "unit": "MILLISECONDS"
+      }
     }
+  }
 }
 ----
+
+Use the pulsar-admin cli to create the connector
+
+[source,shell]
+----
+./bin/pulsar-admin sinks create \
+	--name dse-sink-kv \
+	--classname com.datastax.oss.sink.pulsar.StringCassandraSinkTask \
+	--sink-config-file configs.json \
+	--sink-type cassandra-enhanced
+----
+
+To create the value for the mapping parameter you would provide direction how each column value will be filled. An example string is: `symbol=value.symbol, ts=value.ts, exchange=value.exchange, industry=value.industry, name=key, value=value.value`.