From bb300492a3de592274f35c284b12d3624b1a2e2f Mon Sep 17 00:00:00 2001
From: RickyHuo <huochen1994@163.com>
Date: Fri, 8 Dec 2017 15:17:49 +0800
Subject: [PATCH] add replace_include and replace_exclude

---
 src/main/java/com/sina/bip/hangout/outputs/Clickhouse.java | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/src/main/java/com/sina/bip/hangout/outputs/Clickhouse.java b/src/main/java/com/sina/bip/hangout/outputs/Clickhouse.java
index eada6c2..878b380 100644
--- a/src/main/java/com/sina/bip/hangout/outputs/Clickhouse.java
+++ b/src/main/java/com/sina/bip/hangout/outputs/Clickhouse.java
@@ -76,7 +76,7 @@ protected void prepare() {
 
         if (this.config.containsKey("replace_include_fields") && this.config.containsKey("replace_exclude_fields")) {
             System.out.println("Replace_include_fields and replace_exclude_fields exist at the same time," +
-                    " please use one of them");
+                    " please use one of them.");
             System.exit(1);
         }
 
@@ -136,7 +136,7 @@ protected StringBuilder makeUpSql(List<Map> events) {
                         String fieldValue = e.get(field).toString();
                         if ((this.replace_include_fields != null && this.replace_include_fields.contains(field)) ||
                                 (this.replace_exclude_fields != null && !this.replace_exclude_fields.contains(field))) {
-                            dealWithQuote(fieldValue);
+                            value += "'" + dealWithQuote(fieldValue) + "'";
                         } else {
                             value += "'" + fieldValue + "'";
                         }
@@ -171,6 +171,7 @@ protected void bulkInsert(Map event) throws Exception{
             Connection conn = balanced.getConnection();
             try {
                 conn.createStatement().execute(sqls.toString());
+                conn.close();
             } catch (SQLException e){
                 System.out.println(sqls.toString());
                 System.out.println(e.toString());