You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@samza.apache.org by ca...@apache.org on 2019/06/10 20:59:28 UTC
[samza] branch master updated: SAMZA-2196: [Scala cleanup] Convert
SerializerConfig to Java (#1038)
This is an automated email from the ASF dual-hosted git repository.
cameronlee pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/samza.git
The following commit(s) were added to refs/heads/master by this push:
new 950fdf0 SAMZA-2196: [Scala cleanup] Convert SerializerConfig to Java (#1038)
950fdf0 is described below
commit 950fdf07318a0c141573dcc7563c198c004d1161
Author: Cameron Lee <ca...@linkedin.com>
AuthorDate: Mon Jun 10 13:59:24 2019 -0700
SAMZA-2196: [Scala cleanup] Convert SerializerConfig to Java (#1038)
* SAMZA-2196: [Scala cleanup] Convert SerializerConfig to Java
* rename JavaSerializerConfig to SerializerConfig
* clarify method names, use Class.getName
---
.../apache/samza/config/JavaSerializerConfig.java | 54 ----------
.../org/apache/samza/config/SerializerConfig.java | 109 +++++++++++++++++++++
.../execution/JobNodeConfigurationGenerator.java | 2 +-
.../org/apache/samza/storage/StorageRecovery.java | 15 +--
.../apache/samza/table/TableConfigGenerator.java | 2 +-
.../org/apache/samza/config/SerializerConfig.scala | 77 ---------------
.../apache/samza/container/SamzaContainer.scala | 9 +-
.../reporter/MetricsSnapshotReporterFactory.scala | 6 +-
.../apache/samza/config/TestSerializerConfig.java | 82 ++++++++++++++++
.../TestJobNodeConfigurationGenerator.java | 2 +-
.../org/apache/samza/table/TestTableManager.java | 4 +-
.../apache/samza/config/TestSerializerConfig.scala | 57 -----------
.../samza/storage/kv/RocksDbKeyValueReader.java | 13 +--
.../org/apache/samza/config/Log4jSystemConfig.java | 2 +-
.../apache/samza/logging/log4j/StreamAppender.java | 2 +-
.../org/apache/samza/config/Log4jSystemConfig.java | 2 +-
.../samza/logging/log4j2/StreamAppender.java | 2 +-
17 files changed, 216 insertions(+), 224 deletions(-)
diff --git a/samza-core/src/main/java/org/apache/samza/config/JavaSerializerConfig.java b/samza-core/src/main/java/org/apache/samza/config/JavaSerializerConfig.java
deleted file mode 100644
index 946d4e2..0000000
--- a/samza-core/src/main/java/org/apache/samza/config/JavaSerializerConfig.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * 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.samza.config;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * java version of the SerializerConfig
- */
-public class JavaSerializerConfig extends MapConfig {
- private final static String SERIALIZER_PREFIX = "serializers.registry.%s";
- private final static String SERDE = "serializers.registry.%s.class";
-
- public JavaSerializerConfig(Config config) {
- super(config);
- }
-
- public String getSerdeClass(String name) {
- return get(String.format(SERDE, name), null);
- }
-
- /**
- * Useful for getting individual serializers.
- * @return a list of all serializer names from the config file
- */
- public List<String> getSerdeNames() {
- List<String> results = new ArrayList<String>();
- Config subConfig = subset(String.format(SERIALIZER_PREFIX, ""), true);
- for (String key : subConfig.keySet()) {
- if (key.endsWith(".class")) {
- results.add(key.replace(".class", ""));
- }
- }
- return results;
- }
-}
diff --git a/samza-core/src/main/java/org/apache/samza/config/SerializerConfig.java b/samza-core/src/main/java/org/apache/samza/config/SerializerConfig.java
new file mode 100644
index 0000000..11b6484
--- /dev/null
+++ b/samza-core/src/main/java/org/apache/samza/config/SerializerConfig.java
@@ -0,0 +1,109 @@
+/*
+ * 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.samza.config;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import org.apache.samza.SamzaException;
+import org.apache.samza.serializers.ByteBufferSerdeFactory;
+import org.apache.samza.serializers.ByteSerdeFactory;
+import org.apache.samza.serializers.DoubleSerdeFactory;
+import org.apache.samza.serializers.IntegerSerdeFactory;
+import org.apache.samza.serializers.JsonSerdeFactory;
+import org.apache.samza.serializers.LongSerdeFactory;
+import org.apache.samza.serializers.SerializableSerdeFactory;
+import org.apache.samza.serializers.StringSerdeFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Provides helpers for accessing configs related to serialization and deserialization.
+ */
+public class SerializerConfig extends MapConfig {
+ public static final Logger LOGGER = LoggerFactory.getLogger(SerializerConfig.class);
+
+ public static final String SERIALIZER_PREFIX = "serializers.registry.%s";
+ public static final String SERDE_FACTORY_CLASS = SERIALIZER_PREFIX + ".class";
+ public static final String SERIALIZED_INSTANCE_SUFFIX = ".samza.serialized.instance";
+ public static final String SERDE_SERIALIZED_INSTANCE = SERIALIZER_PREFIX + SERIALIZED_INSTANCE_SUFFIX;
+
+ public SerializerConfig(Config config) {
+ super(config);
+ }
+
+ /**
+ * Returns the pre-defined serde factory class name for the provided serde name. If no pre-defined factory exists,
+ * throws an exception.
+ */
+ public static String getPredefinedSerdeFactoryName(String serdeName) {
+ String serdeFactoryName = doGetPredefinedSerdeFactoryName(serdeName);
+ LOGGER.info("Using default serde {} for serde name {}", serdeFactoryName, serdeName);
+ return serdeFactoryName;
+ }
+
+ /**
+ * Get the name of the serde factory class for {@code name}.
+ * @param name name of the serde in the config
+ * @return serde factory class name for {@code name}; empty if no class was found in the config
+ */
+ public Optional<String> getSerdeFactoryClass(String name) {
+ return Optional.ofNullable(get(String.format(SERDE_FACTORY_CLASS, name), null));
+ }
+
+ /**
+ * Gets all serializer names defined in configuration.
+ * @return a list of all serializer names from the config
+ */
+ public List<String> getSerdeNames() {
+ List<String> results = new ArrayList<String>();
+ Config subConfig = subset(String.format(SERIALIZER_PREFIX, ""), true);
+ for (String key : subConfig.keySet()) {
+ if (key.endsWith(".class")) {
+ results.add(key.replace(".class", ""));
+ }
+ }
+ return results;
+ }
+
+ private static String doGetPredefinedSerdeFactoryName(String serdeName) {
+ switch (serdeName) {
+ case "byte":
+ return ByteSerdeFactory.class.getName();
+ case "bytebuffer":
+ return ByteBufferSerdeFactory.class.getName();
+ case "integer":
+ return IntegerSerdeFactory.class.getName();
+ case "json":
+ return JsonSerdeFactory.class.getName();
+ case "long":
+ return LongSerdeFactory.class.getName();
+ case "serializable":
+ return SerializableSerdeFactory.class.getName();
+ case "string":
+ return StringSerdeFactory.class.getName();
+ case "double":
+ return DoubleSerdeFactory.class.getName();
+ default:
+ throw new SamzaException(String.format("No pre-defined factory class name for serde name %s", serdeName));
+ }
+ }
+}
diff --git a/samza-core/src/main/java/org/apache/samza/execution/JobNodeConfigurationGenerator.java b/samza-core/src/main/java/org/apache/samza/execution/JobNodeConfigurationGenerator.java
index fdcce38..b926dd1 100644
--- a/samza-core/src/main/java/org/apache/samza/execution/JobNodeConfigurationGenerator.java
+++ b/samza-core/src/main/java/org/apache/samza/execution/JobNodeConfigurationGenerator.java
@@ -308,7 +308,7 @@ import org.slf4j.LoggerFactory;
serdes.forEach(serde -> {
String serdeName = serdeUUIDs.computeIfAbsent(serde,
s -> serde.getClass().getSimpleName() + "-" + UUID.randomUUID().toString());
- configs.putIfAbsent(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE(), serdeName),
+ configs.putIfAbsent(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE, serdeName),
base64Encoder.encodeToString(serializableSerde.toBytes(serde)));
});
diff --git a/samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java b/samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java
index b7e8a6e..38a5b9b 100644
--- a/samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java
+++ b/samza-core/src/main/java/org/apache/samza/storage/StorageRecovery.java
@@ -49,12 +49,10 @@ import org.apache.samza.util.Clock;
import org.apache.samza.util.CommandLine;
import org.apache.samza.util.CoordinatorStreamUtil;
import org.apache.samza.util.ReflectionUtil;
-import org.apache.samza.util.ScalaJavaUtil;
import org.apache.samza.util.StreamUtil;
import org.apache.samza.util.SystemClock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import scala.Option;
/**
@@ -189,16 +187,11 @@ public class StorageRecovery extends CommandLine {
SerializerConfig serializerConfig = new SerializerConfig(jobConfig);
// Adding all serdes from factories
- ScalaJavaUtil.toJavaCollection(serializerConfig.getSerdeNames())
- .stream()
+ serializerConfig.getSerdeNames()
.forEach(serdeName -> {
- Option<String> serdeClassName = serializerConfig.getSerdeClass(serdeName);
-
- if (serdeClassName.isEmpty()) {
- serdeClassName = Option.apply(SerializerConfig.getSerdeFactoryName(serdeName));
- }
-
- Serde serde = ReflectionUtil.getObj(getClass().getClassLoader(), serdeClassName.get(), SerdeFactory.class)
+ String serdeClassName = serializerConfig.getSerdeFactoryClass(serdeName)
+ .orElseGet(() -> SerializerConfig.getPredefinedSerdeFactoryName(serdeName));
+ Serde serde = ReflectionUtil.getObj(getClass().getClassLoader(), serdeClassName, SerdeFactory.class)
.getSerde(serdeName, serializerConfig);
serdeMap.put(serdeName, serde);
});
diff --git a/samza-core/src/main/java/org/apache/samza/table/TableConfigGenerator.java b/samza-core/src/main/java/org/apache/samza/table/TableConfigGenerator.java
index 5d41ad6..9e2f279 100644
--- a/samza-core/src/main/java/org/apache/samza/table/TableConfigGenerator.java
+++ b/samza-core/src/main/java/org/apache/samza/table/TableConfigGenerator.java
@@ -90,7 +90,7 @@ public class TableConfigGenerator {
serdes.forEach(serde -> {
String serdeName = serdeUUIDs.computeIfAbsent(serde,
s -> serde.getClass().getSimpleName() + "-" + UUID.randomUUID().toString());
- serdeConfigs.putIfAbsent(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE(), serdeName),
+ serdeConfigs.putIfAbsent(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE, serdeName),
base64Encoder.encodeToString(serializableSerde.toBytes(serde)));
});
diff --git a/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala b/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
deleted file mode 100644
index 240a296..0000000
--- a/samza-core/src/main/scala/org/apache/samza/config/SerializerConfig.scala
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * 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.samza.config
-
-import org.apache.samza.SamzaException
-import org.apache.samza.serializers.ByteBufferSerdeFactory
-import org.apache.samza.serializers.ByteSerdeFactory
-import org.apache.samza.serializers.DoubleSerdeFactory
-import org.apache.samza.serializers.IntegerSerdeFactory
-import org.apache.samza.serializers.JsonSerdeFactory
-import org.apache.samza.serializers.LongSerdeFactory
-import org.apache.samza.serializers.SerializableSerdeFactory
-import org.apache.samza.serializers.StringSerdeFactory
-import org.apache.samza.util.Util.info
-
-import scala.collection.JavaConverters._
-
-object SerializerConfig {
- // serializer config constants
- val SERIALIZER_PREFIX = "serializers.registry.%s"
- val SERDE_FACTORY_CLASS = "serializers.registry.%s.class"
- val SERIALIZED_INSTANCE_SUFFIX = ".samza.serialized.instance"
- val SERDE_SERIALIZED_INSTANCE = SERIALIZER_PREFIX + SERIALIZED_INSTANCE_SUFFIX
-
- implicit def Config2Serializer(config: Config) = new SerializerConfig(config)
-
- /**
- * Returns the pre-defined serde factory class name for the provided serde name. If no pre-defined factory exists,
- * throws an exception.
- */
- def getSerdeFactoryName(serdeName: String) = {
- val serdeFactoryName = serdeName match {
- case "byte" => classOf[ByteSerdeFactory].getCanonicalName
- case "bytebuffer" => classOf[ByteBufferSerdeFactory].getCanonicalName
- case "integer" => classOf[IntegerSerdeFactory].getCanonicalName
- case "json" => classOf[JsonSerdeFactory].getCanonicalName
- case "long" => classOf[LongSerdeFactory].getCanonicalName
- case "serializable" => classOf[SerializableSerdeFactory[java.io.Serializable]].getCanonicalName
- case "string" => classOf[StringSerdeFactory].getCanonicalName
- case "double" => classOf[DoubleSerdeFactory].getCanonicalName
- case _ => throw new SamzaException("No pre-defined factory class name for serde name %s" format serdeName)
- }
- info("Using default serde %s for serde name %s" format (serdeFactoryName, serdeName))
- serdeFactoryName
- }
-}
-
-class SerializerConfig(config: Config) extends ScalaMapConfig(config) {
- def getSerdeClass(name: String) = getOption(SerializerConfig.SERDE_FACTORY_CLASS format name)
-
- /**
- * Returns a list of all serializer names from the config file. Useful for
- * getting individual serializers.
- */
- import SerializerConfig._
- def getSerdeNames() = {
- val subConf = config.subset(SERIALIZER_PREFIX format "", true)
- subConf.asScala.keys.filter(k => k.endsWith(".class")).map(_.replace(".class", ""))
- }
-}
diff --git a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
index 9734cb5..b55f057 100644
--- a/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
+++ b/samza-core/src/main/scala/org/apache/samza/container/SamzaContainer.scala
@@ -33,7 +33,6 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder
import org.apache.samza.checkpoint.{CheckpointListener, OffsetManager, OffsetManagerMetrics}
import org.apache.samza.config.JobConfig.Config2Job
import org.apache.samza.config.MetricsConfig.Config2Metrics
-import org.apache.samza.config.SerializerConfig.Config2Serializer
import org.apache.samza.config.StreamConfig.Config2Stream
import org.apache.samza.config._
import org.apache.samza.container.disk.DiskSpaceMonitor.Listener
@@ -249,10 +248,10 @@ object SamzaContainer extends Logging {
info("Got system producers: %s" format producers.keys)
- val serdesFromFactories = config.getSerdeNames.map(serdeName => {
- val serdeClassName = config
- .getSerdeClass(serdeName)
- .getOrElse(SerializerConfig.getSerdeFactoryName(serdeName))
+ val serializerConfig = new SerializerConfig(config)
+ val serdesFromFactories = serializerConfig.getSerdeNames.asScala.map(serdeName => {
+ val serdeClassName = JavaOptionals.toRichOptional(serializerConfig.getSerdeFactoryClass(serdeName)).toOption
+ .getOrElse(SerializerConfig.getPredefinedSerdeFactoryName(serdeName))
val serde = ReflectionUtil.getObj(classLoader, serdeClassName, classOf[SerdeFactory[Object]])
.getSerde(serdeName, config)
(serdeName, serde)
diff --git a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
index 74ffcc8..d45b27a 100644
--- a/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
+++ b/samza-core/src/main/scala/org/apache/samza/metrics/reporter/MetricsSnapshotReporterFactory.scala
@@ -21,11 +21,10 @@ package org.apache.samza.metrics.reporter
import org.apache.samza.util.{Logging, StreamUtil, Util}
import org.apache.samza.SamzaException
-import org.apache.samza.config.{Config, SystemConfig}
+import org.apache.samza.config.{Config, SerializerConfig, SystemConfig}
import org.apache.samza.config.JobConfig.Config2Job
import org.apache.samza.config.MetricsConfig.Config2Metrics
import org.apache.samza.config.StreamConfig.Config2Stream
-import org.apache.samza.config.SerializerConfig.Config2Serializer
import org.apache.samza.metrics.MetricsReporter
import org.apache.samza.metrics.MetricsReporterFactory
import org.apache.samza.metrics.MetricsRegistryMap
@@ -71,8 +70,9 @@ class MetricsSnapshotReporterFactory extends MetricsReporterFactory with Logging
val streamSerdeName = config.getStreamMsgSerde(systemStream)
val systemSerdeName = JavaOptionals.toRichOptional(systemConfig.getSystemMsgSerde(systemName)).toOption
val serdeName = streamSerdeName.getOrElse(systemSerdeName.getOrElse(null))
+ val serializerConfig = new SerializerConfig(config)
val serde = if (serdeName != null) {
- config.getSerdeClass(serdeName) match {
+ JavaOptionals.toRichOptional(serializerConfig.getSerdeFactoryClass(serdeName)).toOption match {
case Some(serdeClassName) =>
Util.getObj(serdeClassName, classOf[SerdeFactory[MetricsSnapshot]]).getSerde(serdeName, config)
case _ => null
diff --git a/samza-core/src/test/java/org/apache/samza/config/TestSerializerConfig.java b/samza-core/src/test/java/org/apache/samza/config/TestSerializerConfig.java
new file mode 100644
index 0000000..46e3906
--- /dev/null
+++ b/samza-core/src/test/java/org/apache/samza/config/TestSerializerConfig.java
@@ -0,0 +1,82 @@
+/*
+ * 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.samza.config;
+
+import java.util.List;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.samza.SamzaException;
+import org.apache.samza.serializers.ByteBufferSerdeFactory;
+import org.apache.samza.serializers.ByteSerdeFactory;
+import org.apache.samza.serializers.DoubleSerdeFactory;
+import org.apache.samza.serializers.IntegerSerdeFactory;
+import org.apache.samza.serializers.JsonSerdeFactory;
+import org.apache.samza.serializers.LongSerdeFactory;
+import org.apache.samza.serializers.SerializableSerdeFactory;
+import org.apache.samza.serializers.StringSerdeFactory;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+
+public class TestSerializerConfig {
+ private static final String SERDE_NAME = "mySerdeName";
+
+ @Test
+ public void testGetPredefinedSerdeFactoryName() {
+ assertEquals(ByteSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("byte"));
+ assertEquals(ByteBufferSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("bytebuffer"));
+ assertEquals(IntegerSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("integer"));
+ assertEquals(JsonSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("json"));
+ assertEquals(LongSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("long"));
+ assertEquals(SerializableSerdeFactory.class.getName(),
+ SerializerConfig.getPredefinedSerdeFactoryName("serializable"));
+ assertEquals(StringSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("string"));
+ assertEquals(DoubleSerdeFactory.class.getName(), SerializerConfig.getPredefinedSerdeFactoryName("double"));
+ }
+
+ @Test(expected = SamzaException.class)
+ public void testGetSerdeFactoryNameUnknown() {
+ SerializerConfig.getPredefinedSerdeFactoryName("otherName");
+ }
+
+ @Test
+ public void testGetSerdeFactoryClass() {
+ String serdeClassName = "my.class.serde.name";
+ Config config = new MapConfig(
+ ImmutableMap.of(String.format(SerializerConfig.SERDE_FACTORY_CLASS, SERDE_NAME), serdeClassName));
+ assertEquals(serdeClassName, new SerializerConfig(config).getSerdeFactoryClass(SERDE_NAME).get());
+ assertFalse(new SerializerConfig(config).getSerdeFactoryClass("otherSerdeName").isPresent());
+ }
+
+ @Test
+ public void testGetSerdeNames() {
+ String otherSerdeName = "otherSerdeName";
+ String serdeClassName = "my.class.serde.name";
+ Config config = new MapConfig(
+ ImmutableMap.of(String.format(SerializerConfig.SERDE_FACTORY_CLASS, SERDE_NAME), serdeClassName,
+ String.format(SerializerConfig.SERDE_FACTORY_CLASS, otherSerdeName), serdeClassName));
+ List<String> serdeNames = new SerializerConfig(config).getSerdeNames();
+ ImmutableSet<String> expectedSerdeNames = ImmutableSet.of(SERDE_NAME, otherSerdeName);
+ assertEquals(expectedSerdeNames.size(), serdeNames.size()); // ensures no duplicates
+ // can't check ordering in a stable way since values come from the key set of a map, so just check entries
+ assertEquals(expectedSerdeNames, ImmutableSet.copyOf(serdeNames));
+ }
+}
diff --git a/samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java b/samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java
index f60be4f..6f1b9c0 100644
--- a/samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java
+++ b/samza-core/src/test/java/org/apache/samza/execution/TestJobNodeConfigurationGenerator.java
@@ -289,7 +289,7 @@ public class TestJobNodeConfigurationGenerator extends ExecutionPlannerTestBase
SerializableSerde<Serde> serializableSerde = new SerializableSerde<>();
assertEquals(numSerdes, serializers.size());
return serializers.entrySet().stream().collect(Collectors.toMap(
- e -> e.getKey().replace(SerializerConfig.SERIALIZED_INSTANCE_SUFFIX(), ""),
+ e -> e.getKey().replace(SerializerConfig.SERIALIZED_INSTANCE_SUFFIX, ""),
e -> serializableSerde.fromBytes(Base64.getDecoder().decode(e.getValue().getBytes()))
));
}
diff --git a/samza-core/src/test/java/org/apache/samza/table/TestTableManager.java b/samza-core/src/test/java/org/apache/samza/table/TestTableManager.java
index d5de9ea..20e16f4 100644
--- a/samza-core/src/test/java/org/apache/samza/table/TestTableManager.java
+++ b/samza-core/src/test/java/org/apache/samza/table/TestTableManager.java
@@ -108,14 +108,14 @@ public class TestTableManager {
private void addKeySerde(Map<String, String> map) {
String serdeId = "key-serde";
- map.put(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE(), serdeId),
+ map.put(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE, serdeId),
serializeSerde(new IntegerSerde()));
map.put(String.format(JavaTableConfig.STORE_KEY_SERDE, TABLE_ID), serdeId);
}
private void addValueSerde(Map<String, String> map) {
String serdeId = "value-serde";
- map.put(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE(), serdeId),
+ map.put(String.format(SerializerConfig.SERDE_SERIALIZED_INSTANCE, serdeId),
serializeSerde(new StringSerde("UTF-8")));
map.put(String.format(JavaTableConfig.STORE_MSG_SERDE, TABLE_ID), serdeId);
}
diff --git a/samza-core/src/test/scala/org/apache/samza/config/TestSerializerConfig.scala b/samza-core/src/test/scala/org/apache/samza/config/TestSerializerConfig.scala
deleted file mode 100644
index 83c901c..0000000
--- a/samza-core/src/test/scala/org/apache/samza/config/TestSerializerConfig.scala
+++ /dev/null
@@ -1,57 +0,0 @@
-/*
- * 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.samza.config
-
-import org.apache.samza.SamzaException
-import org.apache.samza.serializers.ByteSerdeFactory
-import org.apache.samza.serializers.DoubleSerdeFactory
-import org.apache.samza.serializers.IntegerSerdeFactory
-import org.apache.samza.serializers.JsonSerdeFactory
-import org.apache.samza.serializers.LongSerdeFactory
-import org.apache.samza.config.SerializerConfig.getSerdeFactoryName
-import org.apache.samza.serializers.SerializableSerdeFactory
-import org.apache.samza.serializers.StringSerdeFactory
-import org.junit.Assert.assertEquals
-import org.junit.Assert.assertTrue
-import org.junit.Test
-
-class TestSerializerConfig {
- @Test
- def testGetSerdeFactoryName {
- val config = new MapConfig
- assertEquals(classOf[ByteSerdeFactory].getName, getSerdeFactoryName("byte"))
- assertEquals(classOf[IntegerSerdeFactory].getName, getSerdeFactoryName("integer"))
- assertEquals(classOf[JsonSerdeFactory].getName, getSerdeFactoryName("json"))
- assertEquals(classOf[LongSerdeFactory].getName, getSerdeFactoryName("long"))
- assertEquals(classOf[SerializableSerdeFactory[java.io.Serializable@unchecked]].getName, getSerdeFactoryName("serializable"))
- assertEquals(classOf[StringSerdeFactory].getName, getSerdeFactoryName("string"))
- assertEquals(classOf[DoubleSerdeFactory].getName, getSerdeFactoryName("double"))
-
- // throw SamzaException if can not find the correct serde
- var throwSamzaException = false
- try {
- getSerdeFactoryName("otherName")
- } catch {
- case e: SamzaException => throwSamzaException = true
- case _: Exception =>
- }
- assertTrue(throwSamzaException)
- }
-}
diff --git a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbKeyValueReader.java b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbKeyValueReader.java
index c0b1623..8c6adde 100644
--- a/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbKeyValueReader.java
+++ b/samza-kv-rocksdb/src/main/java/org/apache/samza/storage/kv/RocksDbKeyValueReader.java
@@ -22,8 +22,7 @@ package org.apache.samza.storage.kv;
import java.io.File;
import org.apache.samza.SamzaException;
import org.apache.samza.config.Config;
-import org.apache.samza.config.JavaSerializerConfig;
-import org.apache.samza.config.SerializerConfig$;
+import org.apache.samza.config.SerializerConfig;
import org.apache.samza.config.StorageConfig;
import org.apache.samza.serializers.Serde;
import org.apache.samza.serializers.SerdeFactory;
@@ -56,7 +55,7 @@ public class RocksDbKeyValueReader {
public RocksDbKeyValueReader(String storeName, String dbPath, Config config) {
// get the key serde and value serde from the config
StorageConfig storageConfig = new StorageConfig(config);
- JavaSerializerConfig serializerConfig = new JavaSerializerConfig(config);
+ SerializerConfig serializerConfig = new SerializerConfig(config);
keySerde = getSerdeFromName(storageConfig.getStorageKeySerde(storeName).orElse(null), serializerConfig,
getClass().getClassLoader());
@@ -118,11 +117,9 @@ public class RocksDbKeyValueReader {
* @param serializerConfig serializer config
* @return a Serde of this serde name
*/
- private Serde<Object> getSerdeFromName(String name, JavaSerializerConfig serializerConfig, ClassLoader classLoader) {
- String serdeClassName = serializerConfig.getSerdeClass(name);
- if (serdeClassName == null) {
- serdeClassName = SerializerConfig$.MODULE$.getSerdeFactoryName(name);
- }
+ private Serde<Object> getSerdeFromName(String name, SerializerConfig serializerConfig, ClassLoader classLoader) {
+ String serdeClassName =
+ serializerConfig.getSerdeFactoryClass(name).orElseGet(() -> SerializerConfig.getPredefinedSerdeFactoryName(name));
return ReflectionUtil.getObj(classLoader, serdeClassName, SerdeFactory.class).getSerde(name, serializerConfig);
}
}
diff --git a/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java b/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
index 12c03f6..1e50ebc 100644
--- a/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
+++ b/samza-log4j/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
@@ -77,7 +77,7 @@ public class Log4jSystemConfig extends SystemConfig {
* supplied serde name.
*/
public String getSerdeClass(String name) {
- return get(String.format(SerializerConfig.SERDE_FACTORY_CLASS(), name), null);
+ return get(String.format(SerializerConfig.SERDE_FACTORY_CLASS, name), null);
}
public String getStreamSerdeName(String systemName, String streamName) {
diff --git a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
index 61b7475..eee88fd 100644
--- a/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
+++ b/samza-log4j/src/main/java/org/apache/samza/logging/log4j/StreamAppender.java
@@ -396,7 +396,7 @@ public class StreamAppender extends AppenderSkeleton {
ReflectionUtil.getObj(getClass().getClassLoader(), serdeClass, SerdeFactory.class);
serde = serdeFactory.getSerde(systemName, config);
} else {
- String serdeKey = String.format(SerializerConfig.SERDE_FACTORY_CLASS(), serdeName);
+ String serdeKey = String.format(SerializerConfig.SERDE_FACTORY_CLASS, serdeName);
throw new SamzaException("Can not find serializers class for key '" + serdeName + "'. Please specify " +
serdeKey + " property");
}
diff --git a/samza-log4j2/src/main/java/org/apache/samza/config/Log4jSystemConfig.java b/samza-log4j2/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
index 12c03f6..1e50ebc 100644
--- a/samza-log4j2/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
+++ b/samza-log4j2/src/main/java/org/apache/samza/config/Log4jSystemConfig.java
@@ -77,7 +77,7 @@ public class Log4jSystemConfig extends SystemConfig {
* supplied serde name.
*/
public String getSerdeClass(String name) {
- return get(String.format(SerializerConfig.SERDE_FACTORY_CLASS(), name), null);
+ return get(String.format(SerializerConfig.SERDE_FACTORY_CLASS, name), null);
}
public String getStreamSerdeName(String systemName, String streamName) {
diff --git a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
index b47ddab..28db042 100644
--- a/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
+++ b/samza-log4j2/src/main/java/org/apache/samza/logging/log4j2/StreamAppender.java
@@ -418,7 +418,7 @@ public class StreamAppender extends AbstractAppender {
ReflectionUtil.getObj(getClass().getClassLoader(), serdeClass, SerdeFactory.class);
serde = serdeFactory.getSerde(systemName, config);
} else {
- String serdeKey = String.format(SerializerConfig.SERDE_FACTORY_CLASS(), serdeName);
+ String serdeKey = String.format(SerializerConfig.SERDE_FACTORY_CLASS, serdeName);
throw new SamzaException("Can not find serializers class for key '" + serdeName + "'. Please specify " +
serdeKey + " property");
}