Skip to content

Commit e599d98

Browse files
lawofcycleslawofcycles
committed
Bump commons-codec:commons-codec from 1.21.0 to 1.22.0 in /buildSrc
Signed-off-by: Sotaro Hikita <bering1814@gmail.com>
1 parent 525183e commit e599d98

6 files changed

Lines changed: 367 additions & 1 deletion

File tree

‎buildSrc/build.gradle‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ dependencies {
7373

7474
// Required for dependency licenses task
7575
implementation 'org.apache.rat:apache-rat:0.15'
76-
implementation 'commons-codec:commons-codec:1.21.0'
76+
implementation 'commons-codec:commons-codec:1.22.0'
7777

7878
if (localRepo) {
7979
implementation(name: "build-tools-${buildToolsVersion}") {

‎test-scripts/aoss_bulk_size_test.py‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
from pyspark.sql import SparkSession
2+
from pyspark.sql.types import StructType, StructField, StringType, ArrayType, DoubleType, IntegerType
3+
import json
4+
5+
spark = SparkSession.builder.appName("AOSSBulkSizeReproTest").getOrCreate()
6+
7+
# Generate large records similar to Stacy's data (features array with many nested objects)
8+
def generate_large_record(product_id):
9+
features = []
10+
for i in range(20):
11+
features.append({
12+
"id": f"feature_{i}",
13+
"value": f"value_{i}" * 10,
14+
"numeric_value": float(i) * 1.5,
15+
"data_type": "numeric",
16+
"source": "listing",
17+
"unit": "count",
18+
"narrative": f"This is a long narrative description for feature {i} that simulates the kind of verbose text found in real product data. " * 3,
19+
"positive_count": i,
20+
"negative_count": 0,
21+
"observation_count": i + 1
22+
})
23+
use_cases = [{"id": f"uc_case_{i}", "score": float(i % 4)} for i in range(10)]
24+
customer_segments = [{"id": f"seg_{i}", "score": float(i % 4)} for i in range(8)]
25+
26+
return {
27+
"product_id": product_id,
28+
"title": f"Test Product {product_id} with a very long title that simulates real product data " * 3,
29+
"brand_id": "12345",
30+
"brand_name": "TestBrand",
31+
"product_type": "SHOES",
32+
"price": 99.99,
33+
"ordered_units": 100,
34+
"parent_asin": product_id,
35+
"child_asin_list": ",".join([f"CHILD_{product_id}_{i}" for i in range(10)]),
36+
"classification_description": "Base Product",
37+
"variant_count": 10,
38+
"variant_colors": "Red,Blue,Green,Black,White",
39+
"variant_sizes": "S,M,L,XL,XXL",
40+
"gms_usd": 5000.0,
41+
"features_json": json.dumps(features),
42+
"use_cases_json": json.dumps(use_cases),
43+
"customer_segments_json": json.dumps(customer_segments)
44+
}
45+
46+
# Create 10 records
47+
records = [generate_large_record(f"PROD_{i:04d}") for i in range(10)]
48+
49+
# Create DataFrame
50+
df = spark.createDataFrame(records)
51+
52+
# Print approximate record size
53+
sample = json.dumps(records[0])
54+
print(f"Approximate single record JSON size: {len(sample)} bytes")
55+
print(f"Total records: {len(records)}")
56+
57+
# Write with default batch size (should trigger bulk with multiple docs)
58+
os_config = {
59+
"opensearch.nodes": "https://ppyv4gorhbllttyd2vpi.ap-northeast-1.aoss.amazonaws.com",
60+
"opensearch.port": "443",
61+
"opensearch.net.ssl": "true",
62+
"opensearch.nodes.wan.only": "true",
63+
"opensearch.aws.sigv4.enabled": "true",
64+
"opensearch.aws.sigv4.region": "ap-northeast-1",
65+
"opensearch.aws.sigv4.service.name": "aoss",
66+
"opensearch.serverless": "true",
67+
"opensearch.mapping.id": "product_id",
68+
"opensearch.batch.size.entries": "5",
69+
"opensearch.batch.size.bytes": "10mb"
70+
}
71+
72+
print("Writing with batch.size.entries=5 ...")
73+
df.write.format("org.opensearch.spark.sql") \
74+
.options(**os_config) \
75+
.mode("append") \
76+
.save("bulk_size_test")
77+
78+
print("SUCCESS: Write completed with batch size 5")

‎test-scripts/aoss_mass_test.py‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
from pyspark.sql import SparkSession
2+
import json
3+
4+
spark = SparkSession.builder.appName("AOSSBulkMassTest").getOrCreate()
5+
6+
# Base record from Stacy's data - use as template and vary product_id
7+
base = {"title":"Megadeth Countdown to Extinction Official Tee T-Shirt Mens Unisex (Large) Black","brand_id":"3898204","brand_name":"Rock Off officially licensed products","product_type":"SHIRT","price":0.0,"ordered_units":4,"parent_asin":"B0DVHT64KM","child_asin_list":"B0DVGZXFD2","classification_description":"Base Product","variant_count":1,"variant_colors":"Black","variant_styles":"Classic","variant_sizes":"Large","gms_usd":103.68,"features":[{"id":"cotton_percentage","value":"100.0","numeric_value":100.0,"data_type":"numeric","source":"listing","unit":"percent","positive_count":0,"negative_count":0,"observation_count":0},{"id":"fabric_composition","value":"[cotton]","data_type":"list[categorical]","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"graphic_print","value":"true","data_type":"boolean","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"intended_use","value":"[casual, costume / cosplay]","data_type":"list[categorical]","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"officially_licensed","value":"true","data_type":"boolean","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"pack_quantity","value":"1.0","numeric_value":1.0,"data_type":"numeric","source":"listing","unit":"count","positive_count":0,"negative_count":0,"observation_count":0},{"id":"preshrunk","value":"true","data_type":"boolean","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"shirt_style","value":"t-shirt","data_type":"categorical","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"shrinkage_after_wash","value":"minimal","data_type":"ordinal","source":"listing","ordinal_rank":"0.0","positive_count":0,"negative_count":0,"observation_count":0},{"id":"sleeve_style","value":"short sleeve","data_type":"categorical","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"target_gender","value":"unisex","data_type":"categorical","source":"listing","positive_count":0,"negative_count":0,"observation_count":0},{"id":"wash_method","value":"machine wash","data_type":"categorical","source":"listing","narrative":"Listing specifies a normal 40-degree machine wash.","positive_count":0,"negative_count":0,"observation_count":0}],"use_cases":[{"id":"uc_everyday_casual_wear","score":3.0},{"id":"uc_fandom_self_expression","score":3.0},{"id":"uc_gifting_novelty","score":3.0},{"id":"uc_halloween_costume_shirt","score":1.0}],"customer_segments":[{"id":"seg_comfort_seeker","score":1.0},{"id":"seg_fandom_enthusiast","score":3.0},{"id":"seg_gift_giver","score":3.0}]}
8+
9+
NUM_RECORDS = 10000
10+
11+
rows = []
12+
for i in range(NUM_RECORDS):
13+
row = dict(base)
14+
row["product_id"] = f"PROD_{i:06d}"
15+
row["features"] = json.dumps(base["features"])
16+
row["use_cases"] = json.dumps(base["use_cases"])
17+
row["customer_segments"] = json.dumps(base["customer_segments"])
18+
rows.append(row)
19+
20+
df = spark.createDataFrame(rows)
21+
print(f"Record count: {df.count()}")
22+
23+
os_config = {
24+
"opensearch.nodes": "https://ppyv4gorhbllttyd2vpi.ap-northeast-1.aoss.amazonaws.com",
25+
"opensearch.port": "443",
26+
"opensearch.net.ssl": "true",
27+
"opensearch.nodes.wan.only": "true",
28+
"opensearch.aws.sigv4.enabled": "true",
29+
"opensearch.aws.sigv4.region": "ap-northeast-1",
30+
"opensearch.aws.sigv4.service.name": "aoss",
31+
"opensearch.serverless": "true",
32+
"opensearch.mapping.id": "product_id",
33+
"opensearch.batch.size.entries": "100",
34+
"opensearch.batch.size.bytes": "1mb"
35+
}
36+
37+
print(f"Writing {NUM_RECORDS} records with batch.size.entries=100, batch.size.bytes=1mb ...")
38+
try:
39+
df.write.format("org.opensearch.spark.sql") \
40+
.options(**os_config) \
41+
.mode("append") \
42+
.save("bulk_mass_test")
43+
print(f"SUCCESS: {NUM_RECORDS} records written")
44+
except Exception as e:
45+
print(f"FAILED: {e}")

‎test-scripts/aoss_partial_failure_test.py‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
from pyspark.sql import SparkSession
2+
3+
spark = SparkSession.builder.appName("AOSSPartialFailureReproTest").getOrCreate()
4+
5+
# First, create the index with a strict mapping so we can cause partial failures
6+
# product_id: keyword, price: float
7+
# Then send a batch where some docs have valid price (float) and one has invalid price (string)
8+
# This should cause partial success in _bulk, triggering the retry + signature mismatch
9+
10+
# Record 1: valid
11+
# Record 2: invalid price (string instead of float) - should fail mapping
12+
# Record 3: valid
13+
records = [
14+
{"product_id": "VALID_001", "title": "Valid Product 1", "price": 10.0},
15+
{"product_id": "INVALID_002", "title": "Invalid Product", "price": 20.0},
16+
{"product_id": "VALID_003", "title": "Valid Product 3", "price": 30.0},
17+
]
18+
19+
df = spark.createDataFrame(records)
20+
21+
os_config = {
22+
"opensearch.nodes": "https://ppyv4gorhbllttyd2vpi.ap-northeast-1.aoss.amazonaws.com",
23+
"opensearch.port": "443",
24+
"opensearch.net.ssl": "true",
25+
"opensearch.nodes.wan.only": "true",
26+
"opensearch.aws.sigv4.enabled": "true",
27+
"opensearch.aws.sigv4.region": "ap-northeast-1",
28+
"opensearch.aws.sigv4.service.name": "aoss",
29+
"opensearch.serverless": "true",
30+
"opensearch.mapping.id": "product_id",
31+
"opensearch.batch.size.entries": "3",
32+
"opensearch.batch.size.bytes": "10mb"
33+
}
34+
35+
# Step 1: Write valid docs to establish the mapping (price as double)
36+
print("Step 1: Writing valid docs to establish mapping...")
37+
df.write.format("org.opensearch.spark.sql") \
38+
.options(**os_config) \
39+
.mode("append") \
40+
.save("partial_failure_test")
41+
print("Step 1 done.")
42+
43+
# Step 2: Now write a batch where one doc has a conflicting type for a field
44+
# We'll add a field "extra" as integer in the index, then send a string
45+
records_mixed = [
46+
{"product_id": "BATCH_001", "title": "Good Product 1", "price": 10.0, "extra": "100"},
47+
{"product_id": "BATCH_002", "title": "Good Product 2", "price": 20.0, "extra": "not_a_number"},
48+
{"product_id": "BATCH_003", "title": "Good Product 3", "price": 30.0, "extra": "300"},
49+
]
50+
51+
# Actually, causing mapping conflict with opensearch-spark is tricky because
52+
# Spark enforces schema uniformity. Instead, we can exploit AOSS document size limits
53+
# or use duplicate _id with version conflict.
54+
55+
# Better approach: Write with a mapping that has strict: true, then send a doc with unknown field.
56+
# But we can't control that easily from Spark either.
57+
58+
# Simplest approach: use opensearch.mapping.id pointing to a field,
59+
# then in a second write, use "create" semantics (which fails if doc exists)
60+
# by setting opensearch.write.operation = create
61+
62+
print("Step 2: Writing same IDs with 'create' operation to force version conflict...")
63+
os_config_create = dict(os_config)
64+
os_config_create["opensearch.write.operation"] = "create"
65+
os_config_create["opensearch.batch.size.entries"] = "3"
66+
67+
# Write docs that already exist - first 2 will conflict, last one is new
68+
conflict_records = [
69+
{"product_id": "VALID_001", "title": "Duplicate 1", "price": 99.0},
70+
{"product_id": "VALID_003", "title": "Duplicate 3", "price": 99.0},
71+
{"product_id": "NEW_004", "title": "New Product 4", "price": 40.0},
72+
]
73+
74+
df_conflict = spark.createDataFrame(conflict_records)
75+
76+
try:
77+
df_conflict.write.format("org.opensearch.spark.sql") \
78+
.options(**os_config_create) \
79+
.mode("append") \
80+
.save("partial_failure_test")
81+
print("Step 2 completed (unexpected success)")
82+
except Exception as e:
83+
error_msg = str(e)
84+
if "x-amz-content-sha256" in error_msg:
85+
print(f"REPRODUCED: x-amz-content-sha256 invalid error!")
86+
print(f"Error: {error_msg[:500]}")
87+
elif "version_conflict" in error_msg.lower() or "conflict" in error_msg.lower():
88+
print(f"Got conflict error (but not sha256 issue): {error_msg[:500]}")
89+
else:
90+
print(f"Got different error: {error_msg[:500]}")

‎test-scripts/aoss_sha256_mismatch_test.py‎

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
"""
2+
Reproduce x-amz-content-sha256 invalid by sending a request where
3+
the SHA256 header does not match the actual body.
4+
This simulates what happens in opensearch-hadoop when TrackingBytesArray.remove()
5+
is called after partial bulk success: toInputStream() (used for SHA256) returns
6+
stale bytes while writeTo() (actual HTTP body) returns correct bytes.
7+
"""
8+
import boto3
9+
import hashlib
10+
import requests
11+
from botocore.auth import SigV4Auth
12+
from botocore.awsrequest import AWSRequest
13+
14+
ENDPOINT = "https://ppyv4gorhbllttyd2vpi.ap-northeast-1.aoss.amazonaws.com"
15+
REGION = "ap-northeast-1"
16+
SERVICE = "aoss"
17+
INDEX = "bulk_size_test"
18+
19+
session = boto3.Session()
20+
credentials = session.get_credentials().get_frozen_credentials()
21+
22+
# Simulate the bug: body has 2 entries, but SHA256 is computed from 3 entries (including a "removed" one)
23+
entry1 = '{"index":{"_id":"REMOVED_001"}}\n{"title":"this entry was removed"}\n'
24+
entry2 = '{"index":{"_id":"RETRY_002"}}\n{"title":"this is the retry entry 2"}\n'
25+
entry3 = '{"index":{"_id":"RETRY_003"}}\n{"title":"this is the retry entry 3"}\n'
26+
27+
# What writeTo() sends (correct body, after remove of entry1)
28+
actual_body = entry2 + entry3
29+
30+
# What toInputStream() returns (includes removed entry1 bytes)
31+
stale_body = entry1 + entry2 + entry3
32+
33+
# Compute SHA256 from the STALE body (simulating the bug)
34+
wrong_sha256 = hashlib.sha256(stale_body.encode()).hexdigest()
35+
36+
# Compute SHA256 from the CORRECT body (what it should be)
37+
correct_sha256 = hashlib.sha256(actual_body.encode()).hexdigest()
38+
39+
print(f"Actual body size: {len(actual_body)} bytes")
40+
print(f"Stale body size: {len(stale_body)} bytes")
41+
print(f"Correct SHA256: {correct_sha256}")
42+
print(f"Wrong SHA256 (bug): {wrong_sha256}")
43+
print()
44+
45+
# Send request with WRONG SHA256 (reproducing the bug)
46+
url = f"{ENDPOINT}/{INDEX}/_bulk"
47+
headers = {
48+
"Content-Type": "application/x-ndjson",
49+
"x-amz-content-sha256": wrong_sha256,
50+
}
51+
52+
request = AWSRequest(method="POST", url=url, data=actual_body, headers=headers)
53+
SigV4Auth(credentials, SERVICE, REGION).add_auth(request)
54+
55+
# Override x-amz-content-sha256 with the wrong value AFTER signing
56+
# (SigV4Auth computes the correct one, so we need to replace it)
57+
request.headers["x-amz-content-sha256"] = wrong_sha256
58+
59+
print("Sending request with WRONG x-amz-content-sha256...")
60+
response = requests.post(
61+
url,
62+
data=actual_body,
63+
headers=dict(request.headers),
64+
)
65+
print(f"Status: {response.status_code}")
66+
print(f"Response: {response.text[:500]}")
67+
print()
68+
69+
if response.status_code == 403 and "x-amz-content-sha256" in response.text.lower():
70+
print("REPRODUCED: x-amz-content-sha256 invalid error!")
71+
else:
72+
print(f"Different result: {response.status_code}")
73+
74+
# Control: send with CORRECT SHA256
75+
print("\n--- Control: sending with CORRECT x-amz-content-sha256 ---")
76+
headers2 = {
77+
"Content-Type": "application/x-ndjson",
78+
"x-amz-content-sha256": correct_sha256,
79+
}
80+
request2 = AWSRequest(method="POST", url=url, data=actual_body, headers=headers2)
81+
SigV4Auth(credentials, SERVICE, REGION).add_auth(request2)
82+
83+
response2 = requests.post(
84+
url,
85+
data=actual_body,
86+
headers=dict(request2.headers),
87+
)
88+
print(f"Status: {response2.status_code}")
89+
print(f"Response: {response2.text[:500]}")

0 commit comments

Comments
 (0)