STORM-1845: use UTF-8 instead of default encoding

external/storm-hbase/src/main/java/org/apache/storm/hbase/common/Utils.java

@@ -22,6 +22,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
 
 public class Utils {
     private static final Logger LOG = LoggerFactory.getLogger(Utils.class);
@@ -44,7 +45,7 @@ public static byte[] toBytes(Object obj) {
         if(obj == null) {
           return null;
         } else if(obj instanceof String){
-            return ((String)obj).getBytes();
+            return ((String)obj).getBytes(StandardCharsets.UTF_8);
         } else if (obj instanceof Integer){
             return Bytes.toBytes((Integer) obj);
         } else if (obj instanceof Long){

external/storm-hdfs/src/main/java/org/apache/storm/hdfs/trident/format/DelimitedRecordFormat.java

@@ -20,6 +20,8 @@
 import org.apache.storm.tuple.Fields;
 import org.apache.storm.trident.tuple.TridentTuple;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * RecordFormat implementation that uses field and record delimiters.
  * By default uses a comma (",") as the field delimiter and a
@@ -78,6 +80,6 @@ public byte[] format(TridentTuple tuple) {
             }
         }
         sb.append(this.recordDelimiter);
-        return sb.toString().getBytes();
+        return sb.toString().getBytes(StandardCharsets.UTF_8);
     }
 }

external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/DelimitedRecordHiveMapper.java

@@ -34,6 +34,7 @@
 import java.util.Date;
 import java.text.SimpleDateFormat;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 
 public class DelimitedRecordHiveMapper implements HiveMapper {
     private static final Logger LOG = LoggerFactory.getLogger(DelimitedRecordHiveMapper.class);
@@ -133,7 +134,7 @@ public byte[] mapRecord(TridentTuple tuple) {
                 builder.append(fieldDelimiter);
             }
         }
-        return builder.toString().getBytes();
+        return builder.toString().getBytes(StandardCharsets.UTF_8);
     }
 
     private String getPartitionsByTimeFormat() {

external/storm-hive/src/main/java/org/apache/storm/hive/bolt/mapper/JsonRecordHiveMapper.java

@@ -36,6 +36,7 @@
 import java.util.Date;
 import java.text.SimpleDateFormat;
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 
 public class JsonRecordHiveMapper implements HiveMapper {
     private static final Logger LOG = LoggerFactory.getLogger(DelimitedRecordHiveMapper.class);
@@ -122,7 +123,7 @@ public byte[] mapRecord(TridentTuple tuple) {
                 obj.put(field,tuple.getValueByField(field));
             }
         }
-        return obj.toJSONString().getBytes();
+        return obj.toJSONString().getBytes(StandardCharsets.UTF_8);
     }
 
     private String getPartitionsByTimeFormat() {

external/storm-hive/src/main/java/org/apache/storm/hive/common/HiveWriter.java

@@ -27,6 +27,7 @@
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.nio.charset.StandardCharsets;
 
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hive.hcatalog.streaming.*;
@@ -403,7 +404,7 @@ private byte[] generateRecord(Tuple tuple) {
             buf.append(o);
             buf.append(",");
         }
-        return buf.toString().getBytes();
+        return buf.toString().getBytes(StandardCharsets.UTF_8);
     }
 
     /**

external/storm-redis/src/main/java/org/apache/storm/redis/trident/state/AbstractRedisMapState.java

@@ -32,6 +32,7 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.nio.charset.StandardCharsets;
 
 /**
  * AbstractRedisMapState is base class of any RedisMapState, which implements IBackingMap.
@@ -99,7 +100,7 @@ private List<T> deserializeValues(List<List<Object>> keys, List<String> values)
 		List<T> result = new ArrayList<T>(keys.size());
 		for (String value : values) {
 			if (value != null) {
-				result.add((T) getSerializer().deserialize(value.getBytes()));
+				result.add((T) getSerializer().deserialize(value.getBytes(StandardCharsets.UTF_8)));
 			} else {
 				result.add(null);
 			}