Merge pull request apache#20 from chensj2018/8a-jdbc-sink

8a jdbc sink

Author: hejiang2000

flume-ng-core/src/main/java/org/apache/flume/interceptor/HeaderSplitInterceptor.java

@@ -122,28 +122,26 @@ public void close() {
   @Override
   public Event intercept(Event event) {
     Map<String, String> headers = event.getHeaders();
-    
+
     if (!headers.containsKey(splitKey)) {
       logger.error("{} is not in header", splitKey);
       return event;
     }
-    
+
     Matcher matcher = regex.matcher(headers.get(splitKey));
     if (matcher.find()) {
       for (int group = 0, count = matcher.groupCount(); group < count; group++) {
         int groupIndex = group + 1;
         if (groupIndex > serializers.size()) {
           if (logger.isDebugEnabled()) {
-            logger.debug("Skipping group {} to {} due to missing serializer",
-                group, count);
+            logger.debug("Skipping group {} to {} due to missing serializer", group, count);
           }
           break;
         }
-        headers.put(serializers.get(group),matcher.group(groupIndex));
+        headers.put(serializers.get(group), matcher.group(groupIndex));
       }
-    }
-    else {
-    	logger.warn("no mathch any values");
+    } else {
+      logger.warn("no mathch any values");
     }
     return event;
   }

flume-ng-core/src/main/java/org/apache/flume/sink/gbase/GBase8aSink.java

@@ -18,12 +18,20 @@ Licensed to the Apache Software Foundation (ASF) under one or more
 
 package org.apache.flume.sink.gbase;
 
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.Statement;
+
 import org.apache.flume.Channel;
 import org.apache.flume.Context;
 import org.apache.flume.CounterGroup;
 import org.apache.flume.EventDeliveryException;
 import org.apache.flume.conf.Configurable;
 import org.apache.flume.sink.AbstractSink;
+import org.apache.flume.source.http.HTTPSourceConfigurationConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,9 +47,16 @@ public class GBase8aSink extends AbstractSink implements Configurable {
   private static final Logger logger = LoggerFactory.getLogger(GBase8aSink.class);
 
   private PassiveHttpSink httpSink;
+  private String connectUrl;
+  private String driverClassName;
+  private String userName;
+  private String passWord;
+  private String loadSql;
+  private int loadInterval = 0;
+  private Connection conn = null;
+  private Statement stm = null;
 
   private CounterGroup counterGroup;
-  private int batchSize;
 
   public GBase8aSink() {
     counterGroup = new CounterGroup();
@@ -52,53 +67,135 @@ public GBase8aSink() {
   public void configure(Context context) {
     httpSink.configure(context);
 
-    batchSize = context.getInteger(GBase8aSinkConstants.BATCH_SIZE,
-        GBase8aSinkConstants.DFLT_BATCH_SIZE);
-    logger.debug(this.getName() + " " + "batch size set to " + String.valueOf(batchSize));
-    Preconditions.checkArgument(batchSize > 0, "Batch size must be > 0");
+    connectUrl = context.getString(GBase8aSinkConstants.CONNECTION_STRING);
+    Preconditions.checkArgument(connectUrl != null && connectUrl.trim().length() != 0,
+        "connect url must be a string");
+
+    userName = context.getString(GBase8aSinkConstants.CONNECTION_USERNAME);
+    passWord = context.getString(GBase8aSinkConstants.CONNECTION_PASSWORD);
+    driverClassName = context.getString(GBase8aSinkConstants.CONNECTION_DRIVER_CLASS,
+        GBase8aSinkConstants.DFLT_DRIVER_CLASS);
+    
+    loadInterval = context.getInteger(GBase8aSinkConstants.LOAD_INTERVAL,
+        GBase8aSinkConstants.DFLT_LOAD_INTERVAL);
+    Preconditions.checkArgument(loadInterval >= 0 && loadInterval < 30,
+        "loadInterval must be in [0,30).");
+    
+    loadSql = context.getString(GBase8aSinkConstants.SQL_STRING);
+    Preconditions.checkArgument(loadSql != null && loadSql.trim().length() != 0,
+        "load sql must be a string");
+    try {
+      Integer port = context.getInteger(HTTPSourceConfigurationConstants.CONFIG_PORT);
+      /* 获取本机hostname对应的IP */
+      String IP = InetAddress.getLocalHost().getHostAddress();
+      logger.info("local ip : {}, port : {}", IP, port);
+      loadSql = loadSql.replaceAll("\\$\\{localhost\\}", IP + ":" + port);
+    } catch (UnknownHostException e) {
+      logger.error("Error while get local IP. Exception follows.", e);
+    }
   }
 
   @Override
   public Status process() throws EventDeliveryException {
     Status status = Status.READY;
-
-    /*
-     * TODO 等待合适时间通过 jdbc 通知 8a 来读取数据
-     */
+    
+    try {
+      stm.execute(loadSql);
+      if (stm.getUpdateCount() == 0) {
+        Thread.sleep(loadInterval * 1000);
+      }
+    } catch (Exception e) {
+      logger.error("Error while execute  GBase8a. Exception follows.", e);
+      reconnectGBase8a();
+      status = Status.BACKOFF;
+    }
 
     return status;
   }
 
+  /*
+   * (non-Javadoc)
+   * 
+   * @see java.lang.Object#toString()
+   */
+  @Override
+  public String toString() {
+    return "GBase8aSink [httpSink=" + httpSink + ", connectUrl=" + connectUrl + ", driverClassName="
+        + driverClassName + ", userName=" + userName + ", passWord=" + passWord + ", loadSql="
+        + loadSql + ", conn=" + conn + ", stm=" + stm + ", counterGroup=" + counterGroup + "]";
+  }
+
+  /**
+   * connect to GBase 8a MPP cluster.
+   * 
+   * @author chensj
+   */
+  private void connectGBase8a() {
+    try {
+      Class.forName(driverClassName);
+      conn = DriverManager.getConnection(connectUrl, userName, passWord);
+      stm = conn.createStatement();
+      logger.info("connected GBase8a.");
+    } catch (ClassNotFoundException e) {
+      logger.error("Error while Connecting GBase8a. Exception follows.", e);
+    } catch (SQLException e) {
+      logger.error("Error while getConnection GBase8a. Exception follows.", e);
+    }
+  }
+
+  private void reconnectGBase8a() {
+    disconnectGBase8a();
+    connectGBase8a();
+  }
+
   @Override
   public void start() {
     logger.info("Starting {}...", this);
 
     counterGroup.setName(this.getName());
-    super.start();
+    connectGBase8a();
     httpSink.start();
-
+    super.start();
     logger.info("GBase 8a sink {} started.", getName());
   }
 
   @Override
   public void stop() {
     logger.info("GBase 8a sink {} stopping...", getName());
+    disconnectGBase8a();
 
     httpSink.stop();
     super.stop();
 
     logger.info("GBase 8a sink {} stopped. Event metrics: {}", getName(), counterGroup);
   }
 
-  @Override
-  public String toString() {
-    return "GBase8a " + getName() + " { batchSize: " + batchSize + " }";
+  /**
+   * 
+   */
+  private void disconnectGBase8a() {
+    if (stm != null) {
+      try {
+        if (!stm.isClosed()) {
+          stm.close();
+        }
+      } catch (Exception e) {
+        logger.error("Error while close stm. Exception follows.", e);
+      }
+      if (conn != null) {
+        try {
+          conn.close();
+        } catch (SQLException e) {
+          logger.error("Error while close connect. Exception follows.", e);
+        }
+      }
+    }
+
   }
 
   @Override
   public synchronized void setChannel(Channel channel) {
     httpSink.setChannel(channel);
     super.setChannel(channel);
   }
-
 }

flume-ng-core/src/main/java/org/apache/flume/sink/gbase/GBase8aSinkConstants.java

@@ -30,17 +30,17 @@ public class GBase8aSinkConstants {
 
   public static final String GBASE8A_PREFIX = "gbase8a.";
 
-  /* 给 8a 发送 load data 指令的时间间隔 (单位: 秒) */
-  public static final String INTERVAL = "interval";
-
   /* 8a 连接参数 */
   public static final String CONNECTION_STRING = GBASE8A_PREFIX + "jdbcUrl";
   public static final String CONNECTION_USERNAME = GBASE8A_PREFIX + "username";
   public static final String CONNECTION_PASSWORD = GBASE8A_PREFIX + "password";
-
-  /* 8a 加载语句 (直接设置 SQL, 同时支持自动替换 PassiveHttpSink 服务路径 ${URI}) */
+  public static final String CONNECTION_DRIVER_CLASS = GBASE8A_PREFIX + "driverClass";
+  
+  /* 8a 加载语句 (直接设置 SQL, 同时支持自动替换 ${localhost}为本机ip+http服务的port) */
   public static final String SQL_STRING = GBASE8A_PREFIX + "loadSql";
-
+  /* 8a 加载语句的执行间隔(单位:秒)；只有在flume channel空闲是有效*/
+  public static final String LOAD_INTERVAL = GBASE8A_PREFIX + "loadInterval";
+  
   /* handler 参数 (设置时需要加 "handler." 前缀) */
   public static final String BATCH_SIZE = "batchSize";
   public static final String CHARACTER_ENCODING = "characterEncoding";
@@ -52,6 +52,8 @@ public class GBase8aSinkConstants {
   /* 参数默认值 */
   public static final int DFLT_INTERVAL = 10000; // ms
   public static final String DFLT_HANDLER = "org.apache.flume.sink.gbase.PassiveHttpSinkBlobHandler";
+  public static final String DFLT_DRIVER_CLASS = "com.gbase.jdbc.Driver";
+  public static final int DFLT_LOAD_INTERVAL = 5; // s
 
   /* handler 参数默认值 */
   public static final int DFLT_BATCH_SIZE = 10000; // events

flume-ng-core/src/main/java/org/apache/flume/sink/gbase/PassiveHttpSink.java

@@ -25,6 +25,7 @@ Licensed to the Apache Software Foundation (ASF) under one or more
 import java.util.List;
 import java.util.Map;
 
+import javax.servlet.ServletException;
 import javax.servlet.http.HttpServlet;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
@@ -266,6 +267,26 @@ public void doPost(HttpServletRequest request, HttpServletResponse response)
     public void doGet(HttpServletRequest request, HttpServletResponse response) throws IOException {
       doPost(request, response);
     }
+
+    @Override
+    protected void doHead(HttpServletRequest request, HttpServletResponse response)
+        throws ServletException, IOException {
+      try {
+        handler.handle(request, response);
+      } catch (HTTPBadRequestException ex) {
+        LOG.warn("Received bad request from client. ", ex);
+        response.sendError(HttpServletResponse.SC_BAD_REQUEST,
+            "Bad request from client. " + ex.getMessage());
+        return;
+      } catch (Exception ex) {
+        LOG.warn("Unexpected error while sending events. ", ex);
+        response.sendError(HttpServletResponse.SC_INTERNAL_SERVER_ERROR,
+            "Unexpected error while sending events. " + ex.getMessage());
+        return;
+      }
+
+      response.flushBuffer();
+    }
   }
 
   @Override