From d20d930afa602108a305610b4b301a10dc4ebafe Mon Sep 17 00:00:00 2001
From: Cristian Ferretti <37232625+jcferretti@users.noreply.github.com>
Date: Wed, 8 Dec 2021 10:54:55 -0500
Subject: [PATCH] Fix bugs in avro mappings for ConsumeKafka.py. (#1656)

---
 Integrations/python/deephaven/ConsumeKafka.py            | 8 ++++----
 Integrations/python/deephaven/conversion_utils.py        | 9 ++++-----
 .../io/deephaven/integrations/python/PythonTools.java    | 8 +++++++-
 redpanda/examples/python/kafka-produce-avro.py           | 3 ++-
 4 files changed, 17 insertions(+), 11 deletions(-)

diff --git a/Integrations/python/deephaven/ConsumeKafka.py b/Integrations/python/deephaven/ConsumeKafka.py
index 3a3f3fe18e8..d0c47f37dd1 100644
--- 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
index e695b77810b..37887b8f2c2 100644
--- 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
index 0ea6dfe0262..14c97f72055 100644
--- 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
index 9663913ead7..051b4f9ac9f 100644
--- 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