fix: honor format-specific ingestion mapping kind (#174) by tanmaya-panda1 · Pull Request #183 · Azure/kafka-sink-azure-kusto

Tanmaya Panda (tanmaya-panda1) · 2026-05-25T21:19:09Z

Summary

Fixes #174 — the connector previously hard-coded IngestionMappingKind.CSV for every format that wasn't part of the JSON family or Avro, so PARQUET / ORC / W3CLOGFILE ingestions were rejected SDK-side with:

Wrong ingestion mapping for format 'parquet'; mapping kind should be 'Parquet', but was 'Csv'.

Root cause

KustoSinkTask.getTopicsToIngestionProps had:

} else {
    props.setIngestionMapping(mappingRef, IngestionMappingKind.CSV);
}

…catching every format that wasn't JSON-family or Avro and forcing CSV as the mapping kind, which the Java ingest SDK validator (IngestionProperties.java:350) rejects when the format and mapping kind disagree.

Fix

Replaced the if/else cascade with SDK-driven dispatch using DataFormat.getIngestionMappingKind():

Format	Mapping kind (before)	Mapping kind (after)
CSV / TSV / PSV / SCSV / SOHSV / TXT / TSVE / RAW	CSV	CSV ✓
JSON / SINGLEJSON / MULTIJSON	JSON	JSON ✓
AVRO	AVRO	AVRO ✓
APACHEAVRO	CSV ✗	APACHEAVRO ✓
PARQUET	CSV ✗	PARQUET ✓
ORC	CSV ✗	ORC ✓
W3CLOGFILE	CSV ✗	W3CLOGFILE ✓
SSTREAM	CSV ✗	SSTREAM ✓

The JSON family is still collapsed to MULTIJSON for the writer choice (FileWriter constraint), but the mapping kind is now correctly reported as JSON.

Behavior changes

Correct kind for binary/columnar formats — PARQUET / ORC / W3CLOGFILE / APACHEAVRO / SSTREAM now ingest with their correct mapping kind.
Fail-fast on invalid format — typo'd format values (e.g. parqet) now throw ConfigException listing every supported format instead of silently falling back to CSV. This is intentional: silently coercing an invalid format to CSV masked user errors and ingested data into the wrong shape.

Testing

Unit

11 new unit tests in KustoSinkTaskTest, parameterized across every DataFormat × {withMapping, withoutMapping}, JSON-family aliases, case-insensitivity, and the invalid-format path.
Full suite: 179 tests, 0 failures.

Integration

New parquet end-to-end IT case (bytes-parquet) in KustoSinkIT using a pre-built sample.parquet fixture (5 rows, snappy-compressed).
Schema additions in it-table-setup.kql: parquet_mapping for TBL.
Negative IT (shouldFailWhenIngestionFormatIsInvalid) — registers a connector with format: parqet and asserts the connector/task reaches FAILED state.

End-to-end PPE validation

Validated against ingest-sdkse2etestppe.westus2.kusto.windows.net:

=== TEST A: buggy connector path (mappingKind=CSV for parquet) ===
Reproduced expected failure: IngestionClientException: Wrong ingestion mapping for format 'parquet'; mapping kind should be 'Parquet', but was 'Csv'.

=== TEST B: fixed connector path (mappingKind=PARQUET) ===
Final ingestion status: Succeeded
Row count: 3

Follow-up

A second PR will extend the IT matrix with the remaining text and binary format combinations (TSV / PSV / SCSV / SOHSV / TSVE / TXT / RAW / MULTIJSON / SINGLEJSON / ORC / W3CLOGFILE / APACHEAVRO), plus the new tables/mappings their column shapes require. Scoped separately to keep this PR focused on the fix itself.

The connector previously hard-coded IngestionMappingKind.CSV for every format that was not part of the JSON family or Avro. This caused the Kusto Java ingest SDK to reject PARQUET / ORC / W3CLOGFILE ingestions with 'Wrong ingestion mapping for format <X>; mapping kind should be <X>, but was Csv'. Replace the if/else cascade in KustoSinkTask.getTopicsToIngestionProps with SDK-driven dispatch via DataFormat.getIngestionMappingKind(). The JSON family is still collapsed to MULTIJSON for the writer choice (a FileWriter constraint) but its mapping kind is correctly reported as JSON. Behavior changes: * PARQUET / ORC / W3CLOGFILE / APACHEAVRO / SSTREAM mapping references now produce the correct IngestionMappingKind. * Unknown / typo'd format values now fail fast with ConfigException listing every supported format instead of silently falling back to CSV (previous behavior masked user errors). Tests: * 11 new unit tests in KustoSinkTaskTest, parameterized across every DataFormat x {withMapping, withoutMapping} (179 tests pass). * New parquet end-to-end IT case (bytes-parquet) in KustoSinkIT using a pre-built sample.parquet fixture. * Negative IT case asserting connector reaches FAILED state for an invalid format value. * Fix validated end-to-end against ingest-sdkse2etestppe.westus2.kusto .windows.net (Test A reproduces the bug, Test B confirms the fix Succeeds with row count = 3). Fixes #174 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot

Pull request overview

Fixes format-specific ingestion mapping kind selection by using the Kusto ingest SDK’s DataFormat.getIngestionMappingKind() as the source of truth, preventing SDK-side rejections for non-CSV formats (e.g., PARQUET/ORC/W3CLOGFILE).

Changes:

Replace hard-coded mapping-kind selection with SDK-driven DataFormat → IngestionMappingKind dispatch, and add resolveDataFormat() for case-insensitive parsing + fail-fast on unknown formats.
Add comprehensive unit coverage for getTopicsToIngestionProps() across formats, mappings, JSON-family collapsing, and invalid formats.
Extend integration tests/resources to include a parquet end-to-end case (fixture + KQL mapping) and an invalid-format negative test.

Reviewed changes

Copilot reviewed 7 out of 8 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
src/main/java/com/microsoft/azure/kusto/kafka/connect/sink/KustoSinkTask.java	Uses SDK enum dispatch for mapping kind; adds `resolveDataFormat()` and updates JSON handling.
src/test/java/com/microsoft/azure/kusto/kafka/connect/sink/KustoSinkTaskTest.java	Adds parameterized unit tests covering format→mapping-kind resolution, defaults, and invalid formats.
src/test/java/com/microsoft/azure/kusto/kafka/connect/sink/it/KustoSinkIT.java	Adds invalid-format negative IT and parquet bytes fixture ingestion path.
src/test/java/com/microsoft/azure/kusto/kafka/connect/sink/it/containers/KustoKafkaConnectContainerHelper.java	Adds helper to query connector state for IT assertions.
src/test/resources/it-table-setup.kql	Adds parquet ingestion mapping for the integration table.
src/test/resources/format-samples/README.md	Documents binary fixture purpose and regeneration.
src/test/resources/format-samples/build.py	Adds a script to regenerate the parquet fixture.

        String format = mapping.getFormat();
        String mappingName = mapping.getMapping();
        boolean streamingEnabled = mapping.isStreaming();
-        if (StringUtils.isNotBlank(format) && isDataFormatAnyTypeOfJson(format)) {
-            format = IngestionProperties.DataFormat.JSON.name();
+        if (StringUtils.isNotBlank(format)) {
+            IngestionProperties.DataFormat resolved;
+            try {
+                resolved = IngestionProperties.DataFormat.valueOf(format.trim().toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException ignored) {
+                resolved = null;
+            }
+            if (resolved != null && resolved.isJsonFormat()) {
+                format = IngestionProperties.DataFormat.JSON.name();
+            }


+                        expected.put("vstr", "parquet-row-" + i);
+                        expected.put("vlong", vlong);
+                        expected.put("vtype", "bytes-parquet");
+                        expected.put("vdec", "1.23456789");
+                        expectedRecordsProduced.put(vlong, OBJECT_MAPPER.writeValueAsString(expected));


+            int responseCode = httpResponse.getStatusLine().getStatusCode();
+            if (200 <= responseCode && responseCode <= 300) {
+                try {


                        coordinates.table, dataFormat);
        if (dataFormat.startsWith("bytes")) {
            valueFormat = "org.apache.kafka.connect.converters.ByteArrayConverter";
-            // JSON is written as JSON
+            String suffix = dataFormat.split("-")[1];


Copilot

Pull request overview

Copilot reviewed 7 out of 8 changed files in this pull request and generated 4 comments.

+                        expected.put("vstr", "parquet-row-" + i);
+                        expected.put("vlong", vlong);
+                        expected.put("vtype", "bytes-parquet");
+                        expected.put("vdec", "1.23456789");


+                    } catch (Exception e) {
+                        LOGGER.error("Failed to send parquet fixture to {}", targetTopic, e);
+                    }


+        if (StringUtils.isNotBlank(format)) {
+            IngestionProperties.DataFormat resolved;
+            try {
+                resolved = IngestionProperties.DataFormat.valueOf(format.trim().toUpperCase(Locale.ROOT));
+            } catch (IllegalArgumentException ignored) {
+                resolved = null;
+            }
+            if (resolved != null && resolved.isJsonFormat()) {
+                format = IngestionProperties.DataFormat.JSON.name();
+            }
        }


+        try (CloseableHttpClient httpclient = HttpClients.createDefault();
+                CloseableHttpResponse httpResponse = httpclient.execute(httpget)) {
+            int responseCode = httpResponse.getStatusLine().getStatusCode();
+            if (200 <= responseCode && responseCode <= 300) {


Tanmaya Panda (tanmaya-panda1) temporarily deployed to build May 25, 2026 21:19 — with GitHub Actions Inactive

Tanmaya Panda (tanmaya-panda1) requested a review from Copilot May 26, 2026 14:04

Copilot started reviewing on behalf of Tanmaya Panda (tanmaya-panda1) May 26, 2026 14:04 View session

Copilot AI reviewed May 26, 2026

View reviewed changes

Tanmaya Panda (tanmaya-panda1) requested a review from Copilot May 26, 2026 15:01

Copilot started reviewing on behalf of Tanmaya Panda (tanmaya-panda1) May 26, 2026 15:01 View session

Copilot AI reviewed May 26, 2026

View reviewed changes

Navigation Menu

Uh oh!

fix: honor format-specific ingestion mapping kind (#174) - #183

fix: honor format-specific ingestion mapping kind (#174)#183
Tanmaya Panda (tanmaya-panda1) wants to merge 1 commit into
masterfrom
fix/issue-174-parquet-mapping-kind

Tanmaya Panda (tanmaya-panda1) commented May 25, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Uh oh!

Conversation

Tanmaya Panda (tanmaya-panda1) commented May 25, 2026

Summary

Root cause

Fix

Behavior changes

Testing

Unit

Integration

End-to-end PPE validation

Follow-up

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants