Fix bugs in avro mappings for ConsumeKafka.py. (#1656)

deephaven · Dec 8, 2021 · d20d930 · d20d930
1 parent 6c329b3
commit d20d930
Show file tree

Hide file tree

Showing 4 changed files with 17 additions and 11 deletions.
diff --git a/Integrations/python/deephaven/ConsumeKafka.py b/Integrations/python/deephaven/ConsumeKafka.py
@@ -10,8 +10,8 @@
 
 import deephaven.Types as dh
 
-from deephaven.conversion_utils import _isStr, \
-    _dictToProperties, _dictToMap, IDENTITY
+from deephaven.conversion_utils import \
+    _dictToFun, _dictToMap, _dictToProperties, IDENTITY, _isStr
 
 from deephaven.Types import _jclassFromType
 
@@ -223,12 +223,12 @@ def avro(schema, schema_version:str = None, mapping:dict = None, mapping_only:di
             "'mapping_only' expected, instead got both")
     if mapping is not None:
         have_mapping = True
-        # when providing 'mapping_only', fields names not given are mapped as identity
+        # when providing 'mapping', fields names not given are mapped as identity
         mapping = _dictToFun(mapping, default_value=IDENTITY)
     elif mapping_only is not None:
         have_mapping = True
         # when providing 'mapping_only', fields not given are ignored.
-        mapping = _dictToFun(mapping, default_value=None)
+        mapping = _dictToFun(mapping_only, default_value=None)
     else:
         have_mapping = False
     if _isStr(schema):

diff --git a/Integrations/python/deephaven/conversion_utils.py b/Integrations/python/deephaven/conversion_utils.py
@@ -1107,9 +1107,8 @@ def _seqToSet(s):
     return r
 
 @_passThrough
-def _dictToFun(mapping, default_value):
-    mapping = _dictToMap(d)
+def _dictToFun(dict_mapping, default_value):
+    java_map = _dictToMap(dict_mapping)
     if default_value is IDENTITY:
-        return _python_tools_.functionFromMapWithIdentityDefaults(m)
-    else:
-        return _python_tools_.functionfromMapWithDefault(m, default_value)
+        return _python_tools_.functionFromMapWithIdentityDefaults(java_map)
+    return _python_tools_.functionFromMapWithDefault(java_map, default_value)
diff --git a/Integrations/src/main/java/io/deephaven/integrations/python/PythonTools.java b/Integrations/src/main/java/io/deephaven/integrations/python/PythonTools.java
@@ -15,9 +15,12 @@ public class PythonTools {
      * @return the resulting function
      */
     @SuppressWarnings("unused")
-    public static Function<String, String> functionfromMapWithDefault(
+    public static Function<String, String> functionFromMapWithDefault(
             final Map<String, String> map,
             final String defaultValue) {
+        if (map == null) {
+            throw new IllegalArgumentException("Null map");
+        }
         return (final String key) -> map.getOrDefault(key, defaultValue);
     }
 
@@ -29,6 +32,9 @@ public static Function<String, String> functionfromMapWithDefault(
      */
     @SuppressWarnings("unused")
     public static Function<String, String> functionFromMapWithIdentityDefaults(final Map<String, String> map) {
+        if (map == null) {
+            throw new IllegalArgumentException("Null map");
+        }
         return (final String key) -> map.getOrDefault(key, key);
     }
 }
diff --git a/redpanda/examples/python/kafka-produce-avro.py b/redpanda/examples/python/kafka-produce-avro.py
@@ -4,7 +4,8 @@
 # To run this script, you need confluent-kafka libraries installed.
 # To create a dedicated venv for it, you can do:
 #
-# $ mkdir confluent-kafka; cd confluent-kafka
+# $ cd $SOMEWHERE_YOU_WANT_THIS_TO_LIVE
+# $ mkdir confluent-kafka
 # $ python3 -m venv confluent-kafka
 # $ cd confluent-kafka
 # $ source bin/activate