[feature-#1353][kafka-connector] add value deocder

chunjun-connectors/chunjun-connector-emqx/src/main/java/com/dtstack/chunjun/connector/emqx/converter/EmqxColumnConverter.java

@@ -105,7 +105,7 @@ public MqttMessage toExternal(RowData rowData, MqttMessage output) throws Except
  map = Collections.singletonMap("message", row.getString());
  }
 
- output.setPayload(MapUtil.writeValueAsString(map).getBytes());
+ output.setPayload(MapUtil.writeValueAsBytes(map));
  return output;
  }
 }

chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/converter/KafkaColumnConverter.java

@@ -26,6 +26,7 @@
 import com.dtstack.chunjun.decoder.IDecode;
 import com.dtstack.chunjun.decoder.JsonDecoder;
 import com.dtstack.chunjun.decoder.TextDecoder;
+import com.dtstack.chunjun.decoder.ValueDecoder;
 import com.dtstack.chunjun.element.AbstractBaseColumn;
 import com.dtstack.chunjun.element.ColumnRowData;
 import com.dtstack.chunjun.element.column.BigDecimalColumn;
@@ -60,6 +61,7 @@
 import java.util.stream.Collectors;
 
 import static com.dtstack.chunjun.connector.kafka.option.KafkaOptions.DEFAULT_CODEC;
+import static com.dtstack.chunjun.connector.kafka.option.KafkaOptions.VALUE_CODEC;
 
 /**
  * @author chuixue
@@ -69,7 +71,7 @@
 public class KafkaColumnConverter extends AbstractRowConverter<String, Object, byte[], String> {
 
  /** source kafka msg decode */
- private final IDecode decode;
+ private final IDecode decoder;
  /** kafka Conf */
  private final KafkaConf kafkaConf;
  /** kafka sink out fields */
@@ -78,24 +80,34 @@ public class KafkaColumnConverter extends AbstractRowConverter<String, Object, b
  public KafkaColumnConverter(KafkaConf kafkaConf, List<String> keyTypeList) {
  this.kafkaConf = kafkaConf;
  this.outList = keyTypeList;
- if (DEFAULT_CODEC.defaultValue().equals(kafkaConf.getCodec())) {
- this.decode = new JsonDecoder();
- } else {
- this.decode = new TextDecoder();
+ switch (kafkaConf.getCodec()) {
+ case DEFAULT_CODEC:
+ this.decoder = new JsonDecoder();
+ break;
+ case VALUE_CODEC:
+ this.decoder = new ValueDecoder();
+ break;
+ default:
+ this.decoder = new TextDecoder();
  }
  }
 
  public KafkaColumnConverter(KafkaConf kafkaConf) {
  this.commonConf = this.kafkaConf = kafkaConf;
- if (DEFAULT_CODEC.defaultValue().equals(kafkaConf.getCodec())) {
- this.decode = new JsonDecoder();
- } else {
- this.decode = new TextDecoder();
+ switch (kafkaConf.getCodec()) {
+ case DEFAULT_CODEC:
+ this.decoder = new JsonDecoder();
+ break;
+ case VALUE_CODEC:
+ this.decoder = new ValueDecoder();
+ break;
+ default:
+ this.decoder = new TextDecoder();
  }
 
  // Only json need to extract the fields
  if (!CollectionUtils.isEmpty(kafkaConf.getColumn())
- && DEFAULT_CODEC.defaultValue().equals(kafkaConf.getCodec())) {
+ && DEFAULT_CODEC.equals(kafkaConf.getCodec())) {
  List<String> typeList =
  kafkaConf.getColumn().stream()
  .map(FieldConf::getType)
@@ -110,12 +122,13 @@ public KafkaColumnConverter(KafkaConf kafkaConf) {
 
  @Override
  public RowData toInternal(String input) throws Exception {
- Map<String, Object> map = decode.decode(input);
+ Map<String, Object> map = decoder.decode(input);
  ColumnRowData result;
  if (toInternalConverters == null || toInternalConverters.size() == 0) {
  result = new ColumnRowData(1);
  result.addField(new MapColumn(map));
  } else {
+ // Only json decoder will fill toInternalConverters
  List<FieldConf> fieldConfList = kafkaConf.getColumn();
  result = new ColumnRowData(fieldConfList.size());
  for (int i = 0; i < fieldConfList.size(); i++) {
@@ -181,7 +194,7 @@ public byte[] toExternal(RowData rowData, byte[] output) throws Exception {
  for (int i = 0; i < row.getArity(); i++) {
  values.add(row.getField(i) == null ? "" : row.getField(i).asString());
  }
- map = decode.decode(String.join(",", values));
+ map = decoder.decode(String.join(",", values));
  }
  }
 
@@ -196,7 +209,12 @@ public byte[] toExternal(RowData rowData, byte[] output) throws Exception {
  map = keyPartitionMap;
  }
 
- return MapUtil.writeValueAsString(map).getBytes(StandardCharsets.UTF_8);
+ if (VALUE_CODEC.equals(kafkaConf.getCodec())) {
+ return MapUtil.writeValueAsStringWithoutQuote(map.get(ValueDecoder.KEY_MESSAGE))
+ .getBytes(StandardCharsets.UTF_8);
+ } else {
+ return MapUtil.writeValueAsBytes(map);
+ }
  }
 
  @Override

chunjun-connectors/chunjun-connector-kafka/src/main/java/com/dtstack/chunjun/connector/kafka/option/KafkaOptions.java

@@ -18,18 +18,13 @@
 
 package com.dtstack.chunjun.connector.kafka.option;
 
-import org.apache.flink.configuration.ConfigOption;
-import org.apache.flink.configuration.ConfigOptions;
-
 /**
  * @author chuixue
  * @create 2021-06-07 15:53
  * @description
  */
 public class KafkaOptions {
- public static final ConfigOption<String> DEFAULT_CODEC =
- ConfigOptions.key("default.codec")
- .stringType()
- .defaultValue("json")
- .withDescription("default.codec");
+ public static final String DEFAULT_CODEC = "json";
+
+ public static final String VALUE_CODEC = "value";
 }

chunjun-core/src/main/java/com/dtstack/chunjun/decoder/ValueDecoder.java

@@ -0,0 +1,34 @@
+/*
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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 com.dtstack.chunjun.decoder;
+
+import java.io.Serializable;
+import java.util.Collections;
+import java.util.Map;
+
+public class ValueDecoder implements IDecode, Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ public static final String KEY_MESSAGE = "value";
+
+ @Override
+ public Map<String, Object> decode(final String message) {
+ return Collections.singletonMap(KEY_MESSAGE, message);
+ }
+}

chunjun-core/src/main/java/com/dtstack/chunjun/util/MapUtil.java

@@ -76,8 +76,12 @@ public static <T> T jsonStrToObject(String jsonStr, Class<T> clazz) throws IOExc
  return objectMapper.readValue(jsonStr, clazz);
  }
 
- public static String writeValueAsString(Object obj) throws JsonProcessingException {
- return objectMapper.writeValueAsString(obj);
+ public static String writeValueAsStringWithoutQuote(Object obj) throws JsonProcessingException {
+ return objectMapper.writeValueAsString(obj).replace("\"", "");
+ }
+
+ public static byte[] writeValueAsBytes(Object obj) throws JsonProcessingException {
+ return objectMapper.writeValueAsBytes(obj);
  }
 
  /**

chunjun-core/src/test/java/com/dtstack/chunjun/decoder/ValueDecoderTest.java

@@ -0,0 +1,46 @@
+/*
+ * 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 com.dtstack.chunjun.decoder;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+class ValueDecoderTest {
+
+ private ValueDecoder valueDecoder;
+
+ @BeforeEach
+ void setUp() {
+ valueDecoder = new ValueDecoder();
+ }
+
+ @Test
+ @DisplayName("Should return a map with the message")
+ void decodeShouldReturnAMapWithTheMessage() {
+ final String message = "Hello World";
+ final Map<String, Object> result = valueDecoder.decode(message);
+ assertEquals(Collections.singletonMap("value", message), result);
+ }
+}