Github user QiangCai commented on a diff in the pull request:
https://github.com/apache/carbondata/pull/2544#discussion_r205995493
--- Diff:
store/sql/src/main/java/org/apache/carbondata/dis/DisProducer.java ---
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.carbondata.dis;
+
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+import java.util.Random;
+import java.util.Timer;
+import java.util.TimerTask;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.carbondata.common.logging.LogService;
+import org.apache.carbondata.common.logging.LogServiceFactory;
+
+import com.huaweicloud.dis.DIS;
+import com.huaweicloud.dis.DISClientBuilder;
+import com.huaweicloud.dis.exception.DISClientException;
+import com.huaweicloud.dis.http.exception.ResourceAccessException;
+import com.huaweicloud.dis.iface.data.request.PutRecordsRequest;
+import com.huaweicloud.dis.iface.data.request.PutRecordsRequestEntry;
+import com.huaweicloud.dis.iface.data.response.PutRecordsResult;
+import com.huaweicloud.dis.iface.data.response.PutRecordsResultEntry;
+
+public class DisProducer {
+
+ private static AtomicLong eventId = new AtomicLong(0);
+
+ private static final LogService LOGGER =
+ LogServiceFactory.getLogService(DisProducer.class.getName());
+
+ public static void main(String[] args) {
+ if (args.length < 6) {
+ System.err.println(
+ "Usage: DisProducer <stream name> <endpoint> <region> <ak> <sk>
<project id> ");
+ return;
+ }
+
+ DIS dic =
DISClientBuilder.standard().withEndpoint(args[1]).withAk(args[3]).withSk(args[4])
+ .withProjectId(args[5]).withRegion(args[2]).build();
+
+ Sensor sensor = new Sensor(dic, args[0]);
+ Timer timer = new Timer();
+ timer.schedule(sensor, 0, 5000);
+
+ }
+
+ static class Sensor extends TimerTask {
+ private DIS dic;
+
+ private String streamName;
+
+ private SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd
HH:mm:ss");
+
+ private Random random = new Random();
+
+ private int i = 0;
+ private int flag = 1;
+
+ Sensor(DIS dic, String streamName) {
+ this.dic = dic;
+ this.streamName = streamName;
+ }
+
+ @Override public void run() {
+ uploadData();
+ //recordSensor();
+ }
+
+ private void uploadData() {
+ PutRecordsRequest putRecordsRequest = new PutRecordsRequest();
+ putRecordsRequest.setStreamName(streamName);
+ List<PutRecordsRequestEntry> putRecordsRequestEntryList = new
ArrayList<>();
+ PutRecordsRequestEntry putRecordsRequestEntry = new
PutRecordsRequestEntry();
+ putRecordsRequestEntry.setData(ByteBuffer.wrap(recordSensor()));
+ putRecordsRequestEntry
+
.setPartitionKey(String.valueOf(ThreadLocalRandom.current().nextInt(1000000)));
+ putRecordsRequestEntryList.add(putRecordsRequestEntry);
+ putRecordsRequest.setRecords(putRecordsRequestEntryList);
+
+ LOGGER.info("========== BEGIN PUT ============");
+
+ PutRecordsResult putRecordsResult = null;
+ try {
+ putRecordsResult = dic.putRecords(putRecordsRequest);
+ } catch (DISClientException e) {
+ LOGGER.error(e,
+ "Failed to get a normal response, please check params and
retry." + e.getMessage());
+ } catch (ResourceAccessException e) {
+ LOGGER.error(e, "Failed to access endpoint. " + e.getMessage());
+ } catch (Exception e) {
+ LOGGER.error(e, e.getMessage());
+ }
+
+ if (putRecordsResult != null) {
+ LOGGER.info("Put " + putRecordsResult.getRecords().size() + "
records[" + (
+ putRecordsResult.getRecords().size() -
putRecordsResult.getFailedRecordCount().get())
+ + " successful / " + putRecordsResult.getFailedRecordCount() +
" failed].");
+
+ for (int j = 0; j < putRecordsResult.getRecords().size(); j++) {
+ PutRecordsResultEntry putRecordsRequestEntry1 =
putRecordsResult.getRecords().get(j);
+ if (putRecordsRequestEntry1.getErrorCode() != null) {
+ LOGGER.error("[" + new String(
+ putRecordsRequestEntryList.get(j).getData().array(),
Charset.defaultCharset())
+ + "] put failed, errorCode [" +
putRecordsRequestEntry1.getErrorCode()
+ + "], errorMessage [" +
putRecordsRequestEntry1.getErrorMessage() + "]");
+ } else {
+ LOGGER.info("[" + new String(
--- End diff --
ok
---
