removed custom exception

Author: jruaux

src/main/java/com/redislabs/kafkaconnect/common/RedisEnterpriseConnectorConfigException.java

@@ -20,12 +20,12 @@
 import com.github.jcustenborder.kafka.connect.utils.data.TopicPartitionCounter;
 import com.github.jcustenborder.kafka.connect.utils.jackson.ObjectMapperFactory;
 import com.redislabs.kafkaconnect.RedisEnterpriseSinkConnector;
-import com.redislabs.kafkaconnect.common.RedisEnterpriseConnectorConfigException;
 import io.lettuce.core.KeyValue;
 import io.lettuce.core.RedisClient;
 import io.lettuce.core.api.StatefulRedisConnection;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.data.Field;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.errors.ConnectException;
@@ -123,7 +123,7 @@ private OperationItemWriter.RedisOperation<SinkRecord> operation() {
             case ZSET:
                 return new Zadd<>(this::collectionKey, this::key, new ConstantPredicate<>(false), new NullValuePredicate<>(this::score), this::score);
             default:
-                throw new RedisEnterpriseConnectorConfigException("Data structure not supported: " + config.getType());
+                throw new ConfigException(RedisEnterpriseSinkConfig.TYPE, config.getType());
         }
     }
 

src/main/java/com/redislabs/kafkaconnect/sink/RedisEnterpriseSinkTask.java

@@ -200,13 +200,12 @@ public void putZset(RedisServer redis) {
         List<SinkRecord> records = new ArrayList<>(count);
         for (int i = 0; i < count; i++) {
             String value = "zsetmember:" + i;
-            Double score = (double) i;
-            expected.add(ScoredValue.just(score, value));
-            records.add(SinkRecordHelper.write(topic, new SchemaAndValue(Schema.STRING_SCHEMA, value), new SchemaAndValue(Schema.FLOAT64_SCHEMA, score)));
+            expected.add(ScoredValue.just(i, value));
+            records.add(SinkRecordHelper.write(topic, new SchemaAndValue(Schema.STRING_SCHEMA, value), new SchemaAndValue(Schema.FLOAT64_SCHEMA, i)));
         }
         put(topic, RedisEnterpriseSinkConfig.DataType.ZSET, redis, records);
         List<ScoredValue<String>> actual = syncSortedSet(redis).zrangeWithScores(topic, 0, -1);
-        Collections.sort(expected, Comparator.comparing(ScoredValue::getScore));
+        expected.sort(Comparator.comparing(ScoredValue::getScore));
         assertEquals(expected, actual);
     }