fix astra debug log

Author: zzzming

cassandra-astra/README.md

@@ -35,9 +35,9 @@ ASTRA_DB_PASSWORD=
 ```
 Additionally, `keyspace` and `table name` are mandatory to add rows into Astra Cassandra database.
 
-`table schema` is optional, however, the table schema has to be created on the table before any row can be added if the schema is not specified in the configuration.
+`table schema` is optional. However, the table schema has to be created on the table before any row can be added if the schema is not specified in the configuration.
 
-These configuration must be specifed in [the configuration yaml](./config/pulsar-s3-io.yaml)
+These configuration must be specifed in [the configuration yaml](./config/pulsar-astra-sink.yaml)
 
 When set `logLevel: "debug"`, debug logs will be printed by the sink.
 

cassandra-astra/src/main/java/com/kesque/pulsar/sink/cassandra/astra/AstraConfig.java

@@ -85,16 +85,18 @@ public void setBatchSize(int batchSize) {
         this.batchSize = batchSize;
     }
 
-    private boolean isDebug = false;
-    public boolean debugLoglevel() {
-        return isDebug;
-    }
-
     // currently only support debug level
     private String logLevel = "";
     public String getLogLevel() {
         return this.logLevel;
     }
+    public void setLogLevel(String loglevel) {
+        this.logLevel = loglevel;
+    }
+
+    public boolean debugLoglevel() {
+        return this.logLevel.equalsIgnoreCase("debug");
+    }
 
     public static AstraConfig load(String yamlFile) throws IOException {
         ObjectMapper mapper = new ObjectMapper(new YAMLFactory());

cassandra-astra/src/main/java/com/kesque/pulsar/sink/cassandra/astra/AstraSink.java

@@ -72,8 +72,16 @@ public class AstraSink implements Sink<byte[]> {
     */
     @Override
     public void write(Record<byte[]> record) throws Exception {
-        //TODO: toAstra can be its own thread
-        toAstra(record);
+        //TODO: this can be its own thread
+        try {
+            addRow(buildColumnData(record), this.table);
+            record.ack();
+            if (astraConfig.debugLoglevel()) {
+                log.info("successfully add row");
+            }
+        } catch (Exception e) {
+            log.error("failed to send to astra ", e);
+        }
     }
 
     @Override
@@ -194,23 +202,13 @@ private String buildColumnData(Record<byte[]> record) {
         JSONObject obj = new JSONObject(new String(record.getValue()));
         String objStr = "{\"columns\":[";
         for (String key : obj.keySet()) {
-          objStr = objStr + "{\"name\":\"" + key + "\",\"value\":\"" + obj.getString(key) + "\"},";
+            objStr = objStr + "{\"name\":\"" + key + "\",\"value\":\"" + obj.getString(key) + "\"},";
         }
 
         // remove the last comma
         return StringUtils.substring(objStr, 0, objStr.length() - 1) + "]}";
     }
 
-    private void toAstra(Record<byte[]> record) {
-        try {
-            addRow(buildColumnData(record), this.table);
-        } catch (Exception e) {
-            log.error("failed to send to astra ", e);
-        }
-
-        record.ack();
-    }
-
     private String genUUID() {
         return UUID.randomUUID().toString();
     }