added Avro support for KafkaAdapter

CMakeLists.txt

@@ -165,7 +165,7 @@ if(WIN32)
         add_definitions(-DNDEBUG)
     endif()
 
-    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /EHsc /MP /bigobj")
+    set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /EHsc /MP /bigobj /utf-8")
     foreach(warning 4244 4251 4267 4275 4290 4786 4305 4996)
         SET(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} /wd${warning}")
     endforeach(warning)

conda/dev-environment-unix.yml

@@ -4,6 +4,7 @@ channels:
   - nodefaults
 dependencies:
  - astor
+ - avrocpp
  - bison
  - brotli
  - bump-my-version

conda/dev-environment-win.yml

@@ -3,8 +3,9 @@ channels:
   - conda-forge
   - nodefaults
 dependencies:
-#  - bison  # not available on windows
  - astor
+ - avrocpp
+#  - bison  # not available on windows
  - brotli
  - bump-my-version
  - cmake

cpp/cmake/modules/FindAvro.cmake

@@ -0,0 +1,109 @@
+find_path(Avro_INCLUDE_DIR NAMES avro/Encoder.hh)
+find_library(Avro_LIBRARY NAMES avrocpp libavrocpp)
+
+# =============================================================================
+# Workaround for conda-forge avro-cpp fmt::formatter incompatibility on Windows
+# =============================================================================
+# conda-forge's avro-cpp has fmt::formatter specializations with non-const
+# format() methods, but fmt v12+ requires const. This causes MSVC error C2766.
+#
+# Solution: On Windows, we check if the avro headers have this bug, and if so,
+# we create patched versions in the build directory and prepend them to the
+# include path so they shadow the broken originals.
+#
+# This workaround can be removed once conda-forge updates avro-cpp.
+# =============================================================================
+
+set(CSP_AVRO_PATCHED_INCLUDE_DIR "")
+
+if(WIN32 AND Avro_INCLUDE_DIR AND NOT CSP_USE_VCPKG)
+    # Check if avro/Node.hh has the non-const format() bug
+    # The buggy pattern is: "auto format(...) {" without "const" before the brace
+    set(_avro_node_hh "${Avro_INCLUDE_DIR}/avro/Node.hh")
+    set(_avro_types_hh "${Avro_INCLUDE_DIR}/avro/Types.hh")
+    set(_needs_patching FALSE)
+
+    if(EXISTS "${_avro_node_hh}")
+        file(READ "${_avro_node_hh}" _node_hh_content)
+
+        # Check if the file contains fmt::formatter and non-const format()
+        # We look for "auto format" followed by ")" then whitespace then "{"
+        # without "const" in between
+        string(FIND "${_node_hh_content}" "fmt::formatter<avro::Name>" _has_formatter)
+        if(NOT _has_formatter EQUAL -1)
+            # Check specifically for the non-const pattern
+            # Buggy: auto format(const avro::Name &n, FormatContext &ctx) {
+            # Fixed: auto format(const avro::Name &n, FormatContext &ctx) const {
+            string(REGEX MATCH "auto format\\(const avro::Name[^)]+\\)[^c]*\\{" _buggy_pattern "${_node_hh_content}")
+            if(_buggy_pattern)
+                set(_needs_patching TRUE)
+            endif()
+        endif()
+    endif()
+
+    if(_needs_patching)
+        message(STATUS "Detected avro-cpp with non-const fmt::formatter bug - applying build-time patch")
+
+        # Create patched headers directory structure
+        set(CSP_AVRO_PATCHED_INCLUDE_DIR "${CMAKE_BINARY_DIR}/_patched_avro_headers")
+        set(_patched_avro_dir "${CSP_AVRO_PATCHED_INCLUDE_DIR}/avro")
+        file(MAKE_DIRECTORY "${_patched_avro_dir}")
+
+        # Patch Node.hh
+        # Replace: auto format(const avro::Name &n, FormatContext &ctx) {
+        # With:    auto format(const avro::Name &n, FormatContext &ctx) const {
+        string(REGEX REPLACE
+            "(auto format\\(const avro::Name[^)]+\\))[ \t\r\n]*(\\{)"
+            "\\1 const \\2"
+            _patched_node_content
+            "${_node_hh_content}"
+        )
+        file(WRITE "${_patched_avro_dir}/Node.hh" "${_patched_node_content}")
+        message(STATUS "  Patched: avro/Node.hh -> ${_patched_avro_dir}/Node.hh")
+
+        # Patch Types.hh if it exists and has the same bug
+        if(EXISTS "${_avro_types_hh}")
+            file(READ "${_avro_types_hh}" _types_hh_content)
+            string(FIND "${_types_hh_content}" "fmt::formatter<avro::Type>" _has_type_formatter)
+            if(NOT _has_type_formatter EQUAL -1)
+                string(REGEX MATCH "auto format\\(avro::Type[^)]+\\)[^c]*\\{" _types_buggy "${_types_hh_content}")
+                if(_types_buggy)
+                    string(REGEX REPLACE
+                        "(auto format\\(avro::Type[^)]+\\))[ \t\r\n]*(\\{)"
+                        "\\1 const \\2"
+                        _patched_types_content
+                        "${_types_hh_content}"
+                    )
+                    file(WRITE "${_patched_avro_dir}/Types.hh" "${_patched_types_content}")
+                    message(STATUS "  Patched: avro/Types.hh -> ${_patched_avro_dir}/Types.hh")
+                endif()
+            endif()
+        endif()
+    endif()
+endif()
+
+if (NOT TARGET Avro::avrocpp)
+  add_library(Avro::avrocpp SHARED IMPORTED)
+
+  # On Windows, IMPORTED_IMPLIB is the .lib file, IMPORTED_LOCATION is the .dll
+  # On other platforms, IMPORTED_LOCATION is the shared library
+  if(WIN32)
+    set_property(TARGET Avro::avrocpp PROPERTY IMPORTED_IMPLIB "${Avro_LIBRARY}")
+  else()
+    set_property(TARGET Avro::avrocpp PROPERTY IMPORTED_LOCATION "${Avro_LIBRARY}")
+  endif()
+
+  # If we have patched headers, prepend them to include path so they shadow originals
+  if(CSP_AVRO_PATCHED_INCLUDE_DIR)
+    target_include_directories(Avro::avrocpp INTERFACE
+        ${CSP_AVRO_PATCHED_INCLUDE_DIR}  # Patched headers first (shadows originals)
+        ${Avro_INCLUDE_DIR}               # Original headers for everything else
+    )
+  else()
+    target_include_directories(Avro::avrocpp INTERFACE ${Avro_INCLUDE_DIR})
+  endif()
+endif()
+
+include(FindPackageHandleStandardArgs)
+find_package_handle_standard_args(Avro DEFAULT_MSG Avro_LIBRARY Avro_INCLUDE_DIR)
+mark_as_advanced(Avro_INCLUDE_DIR Avro_LIBRARY Avro::avrocpp)

cpp/cmake/modules/FindDepsKafkaAdapter.cmake

@@ -2,11 +2,13 @@ cmake_minimum_required(VERSION 3.7.2)
 
 if (CSP_USE_VCPKG)
   find_package(RdKafka CONFIG REQUIRED)
-  if(NOT WIN32) 
+  find_package(unofficial-avro-cpp CONFIG REQUIRED)
+  if(NOT WIN32)
     # Bad, but a temporary workaround for
     # https://github.com/microsoft/vcpkg/issues/40320
     link_directories(${VCPKG_INSTALLED_DIR}/${VCPKG_TARGET_TRIPLET}/lib)
   endif()
 else()
   find_package(RdKafka REQUIRED)
+  find_package(Avro REQUIRED)
 endif()

cpp/csp/adapters/kafka/CMakeLists.txt

@@ -24,6 +24,13 @@ find_package(DepsKafkaAdapter REQUIRED)
 
 target_link_libraries(csp_kafka_adapter PUBLIC csp_adapter_utils RdKafka::rdkafka RdKafka::rdkafka++)
 
+# Link Avro library
+if(CSP_USE_VCPKG)
+  target_link_libraries(csp_kafka_adapter PUBLIC unofficial::avro-cpp::avrocpp)
+else()
+  target_link_libraries(csp_kafka_adapter PUBLIC Avro::avrocpp)
+endif()
+
 install(TARGETS csp_kafka_adapter
         PUBLIC_HEADER DESTINATION include/csp/adapters/kafka
         RUNTIME DESTINATION ${CSP_RUNTIME_INSTALL_SUBDIR}

cpp/csp/adapters/kafka/KafkaPublisher.cpp

@@ -3,6 +3,7 @@
 #include <csp/adapters/kafka/KafkaPublisher.h>
 #include <csp/adapters/utils/MessageWriter.h>
 #include <csp/adapters/utils/JSONMessageWriter.h>
+#include <csp/adapters/utils/AvroMessageWriter.h>
 
 #include <librdkafka/rdkafkacpp.h>
 
@@ -17,6 +18,8 @@ KafkaPublisher::KafkaPublisher( KafkaAdapterManager * mgr, const Dictionary & pr
     auto protocol = properties.get<std::string>( "protocol" );
     if( protocol == "JSON" )
         m_msgWriter = std::make_shared<utils::JSONMessageWriter>( properties );
+    else if( protocol == "AVRO" )
+        m_msgWriter = std::make_shared<utils::AvroMessageWriter>( properties );
     else if( protocol != "RAW_BYTES" )
         CSP_THROW( NotImplemented, "msg protocol " << protocol << " not currently supported for kafka output adapters" );
 }

cpp/csp/adapters/utils/AvroIncludes.h

@@ -0,0 +1,18 @@
+#ifndef _IN_CSP_ADAPTERS_UTILS_AVROINCLUDES_H
+#define _IN_CSP_ADAPTERS_UTILS_AVROINCLUDES_H
+
+// Centralized avro includes.
+// On Windows conda-forge builds, headers are patched at CMake configure time
+// (see FindAvro.cmake) to fix fmt v12 compatibility.
+
+#include <avro/Compiler.hh>
+#include <avro/Decoder.hh>
+#include <avro/Encoder.hh>
+#include <avro/Generic.hh>
+#include <avro/GenericDatum.hh>
+#include <avro/Node.hh>
+#include <avro/Schema.hh>
+#include <avro/Stream.hh>
+#include <avro/ValidSchema.hh>
+
+#endif