Skip to content

Commit b6fc8ee

Browse files
author
markheger
authored
Merge pull request #195 from IBMStreams/develop
2.0.2
2 parents 11a5fe1 + 3a37fbe commit b6fc8ee

11 files changed

Lines changed: 1160 additions & 43 deletions

File tree

‎com.ibm.streamsx.objectstorage/impl/java/src/com/ibm/streamsx/objectstorage/BaseObjectStorageSink.java‎

Lines changed: 83 additions & 32 deletions
Original file line numberDiff line numberDiff line change
@@ -83,6 +83,7 @@ public class BaseObjectStorageSink extends AbstractObjectStorageOperator impleme
8383
public static final String ACTIVE_OBJECTS_METRIC = "nActiveObjects";
8484
public static final String CLOSED_OBJECTS_METRIC = "nClosedObjects";
8585
public static final String CLOSE_FAILURES_METRIC = "nCloseFailures";
86+
public static final String WRITE_FAILURES_METRIC = "nWriteFailures";
8687
public static final String EXPIRED_OBJECTS_METRIC = "nExpiredObjects";
8788
public static final String EVICTED_OBJECTS_METRIC = "nEvictedObjects";
8889
public static final String MAX_CONCURRENT_PARTITIONS_NUM_METRIC = "maxConcurrentPartitionsNum";
@@ -185,6 +186,7 @@ public class BaseObjectStorageSink extends AbstractObjectStorageOperator impleme
185186
private Metric nActiveObjects;
186187
private Metric nClosedObjects;
187188
private Metric nCloseFailures;
189+
private Metric nWriteFailures;
188190
private Metric nExpiredObjects;
189191
private Metric nEvictedObjects;
190192
private Metric startupTimeMillisecs;
@@ -222,6 +224,11 @@ public void setnCloseFailures (Metric nCloseFailures) {
222224
this.nCloseFailures = nCloseFailures;
223225
}
224226

227+
@CustomMetric (kind = Metric.Kind.COUNTER, name = "nWriteFailures", description = "Number of failures during create object or write to object")
228+
public void setnWriteFailures (Metric nWriteFailures) {
229+
this.nWriteFailures = nWriteFailures;
230+
}
231+
225232
@CustomMetric (kind = Metric.Kind.GAUGE, name = "objectSizeMin", description = "Minimal size of an object uploaded to COS in bytes.")
226233
public void setobjectSizeMin (Metric objectSizeMin) {
227234
this.objectSizeMin = objectSizeMin;
@@ -1227,6 +1234,10 @@ public Metric getCloseObjectsMetric() {
12271234
public Metric getCloseFailuresMetric() {
12281235
return nCloseFailures;
12291236
}
1237+
1238+
public Metric getWriteFailuresMetric() {
1239+
return nWriteFailures;
1240+
}
12301241

12311242
public Metric getExpiredObjectsMetric() {
12321243
return nExpiredObjects;
@@ -1359,22 +1370,35 @@ private void createObject(String partitionPath, String objectname, boolean isWri
13591370
TRACE.log(TraceLevel.INFO, "Create Object '" + objectname + "' with storage format '" + getStorageFormat() + "'");
13601371
}
13611372

1362-
// create new OS object
1363-
// if partitioning required - create object in the proper partition
1364-
if (isWritable) {
1365-
fObjectToWrite = fOSObjectFactory.createWritableObject(partitionPath, objectname, fHeaderRow, fDataIndex, fDataType, getObjectStorageClient(), this.parquetSchemaStr, this.parquetWriterConfig);
1366-
} else {
1367-
fObjectToWrite = fOSObjectFactory.createObject(partitionPath, objectname, fHeaderRow, fDataIndex, fDataType);
1368-
}
1369-
1370-
if ((TRACE.isLoggable(TraceLevel.TRACE)) && (isParquetPartitioned())) {
1371-
TRACE.log(TraceLevel.TRACE, "Register Object '" + objectname + "' in partition registry using partition key '" + fObjectToWrite.getPartitionPath() + "'");
1373+
try {
1374+
// create new OS object
1375+
// if partitioning required - create object in the proper partition
1376+
if (isWritable) {
1377+
fObjectToWrite = fOSObjectFactory.createWritableObject(partitionPath, objectname, fHeaderRow, fDataIndex, fDataType, getObjectStorageClient(), this.parquetSchemaStr, this.parquetWriterConfig);
1378+
} else {
1379+
fObjectToWrite = fOSObjectFactory.createObject(partitionPath, objectname, fHeaderRow, fDataIndex, fDataType);
1380+
}
1381+
1382+
if ((TRACE.isLoggable(TraceLevel.TRACE)) && (isParquetPartitioned())) {
1383+
TRACE.log(TraceLevel.TRACE, "Register Object '" + objectname + "' in partition registry using partition key '" + fObjectToWrite.getPartitionPath() + "'");
1384+
}
1385+
1386+
// in the OS objects registry
1387+
fOSObjectRegistry.register(fObjectToWrite.getPartitionPath(), fObjectToWrite);
1388+
if ((isConsistentRegion()) && (isParquetPartitioned)) {
1389+
partitionedPathList.add(fObjectToWrite.getPartitionPath()); // store object name for here, used by deleteOnReset
1390+
}
13721391
}
1373-
1374-
// in the OS objects registry
1375-
fOSObjectRegistry.register(fObjectToWrite.getPartitionPath(), fObjectToWrite);
1376-
if ((isConsistentRegion()) && (isParquetPartitioned)) {
1377-
partitionedPathList.add(fObjectToWrite.getPartitionPath()); // store object name for here, used by deleteOnReset
1392+
catch (Exception e) {
1393+
String errRootCause = com.ibm.streamsx.objectstorage.Utils.getErrorRootCause(e);
1394+
String errMsg = Messages.getString("OBJECTSTORAGE_SOURCE_NOT_OPENING_OBJECT", errRootCause);
1395+
TRACE.log(TraceLevel.ERROR, errMsg);
1396+
TRACE.log(TraceLevel.ERROR, "Failed create object: "+objectname+" - Exception: " + e.getMessage());
1397+
LOGGER.log(TraceLevel.ERROR, errMsg);
1398+
getWriteFailuresMetric().increment();
1399+
if (isConsistentRegion()) {
1400+
throw e;
1401+
}
13781402
}
13791403
}
13801404

@@ -1458,13 +1482,33 @@ public void processPunctuation(StreamingInput<Tuple> arg0, Punctuation punct)
14581482
fOSObjectRegistry.closeAll();
14591483
}
14601484
else {
1461-
// close synchronously
1462-
fOSObjectRegistry.closeAllImmediately();
1485+
try {
1486+
// close synchronously
1487+
fOSObjectRegistry.closeAllImmediately();
1488+
}
1489+
catch (Exception e) {
1490+
String errRootCause = com.ibm.streamsx.objectstorage.Utils.getErrorRootCause(e);
1491+
String errMsg = Messages.getString("OBJECTSTORAGE_SINK_OBJECT_CLOSE_FAILURE", getBucketName(), currentObjectName, errRootCause);
1492+
TRACE.log(TraceLevel.ERROR, errMsg);
1493+
TRACE.log(TraceLevel.ERROR, "Failed to close on window punctuation marker. Exception: " + e.getMessage());
1494+
LOGGER.log(TraceLevel.ERROR, errMsg);
1495+
getActiveObjectsMetric().setValue(0);
1496+
getCloseFailuresMetric().increment();
1497+
}
14631498
}
14641499
}
14651500
else if (punct == Punctuation.FINAL_MARKER) {
1466-
// close synchronously
1467-
fOSObjectRegistry.closeAllImmediately();
1501+
try {
1502+
// close synchronously
1503+
fOSObjectRegistry.closeAllImmediately();
1504+
}
1505+
catch (Exception e) {
1506+
String errRootCause = com.ibm.streamsx.objectstorage.Utils.getErrorRootCause(e);
1507+
String errMsg = Messages.getString("OBJECTSTORAGE_SINK_OBJECT_CLOSE_FAILURE", getBucketName(), currentObjectName, errRootCause);
1508+
TRACE.log(TraceLevel.ERROR, errMsg);
1509+
TRACE.log(TraceLevel.ERROR, "Failed to close on final punctuation marker. Exception: " + e.getMessage());
1510+
LOGGER.log(TraceLevel.ERROR, errMsg);
1511+
}
14681512
super.processPunctuation(arg0, punct);
14691513
}
14701514
}
@@ -1546,24 +1590,31 @@ synchronized public void process(StreamingInput<Tuple> stream, Tuple tuple)
15461590
createObject(partitionKey, currentObjectName);
15471591

15481592
if (TRACE.isLoggable(TraceLevel.TRACE)) {
1549-
TRACE.log(TraceLevel.TRACE, "New object '" + fObjectToWrite.getPath() + "' has been created for partition key '" + partitionKey + "'");
1593+
if (fObjectToWrite != null) {
1594+
TRACE.log(TraceLevel.TRACE, "New object '" + fObjectToWrite.getPath() + "' has been created for partition key '" + partitionKey + "'");
1595+
}
15501596
}
15511597
}
15521598

1553-
try {
1554-
fObjectToWrite.writeTuple(tuple, partitionKey, fOSObjectRegistry);
1555-
// update metrics
1556-
updateCachedDataMetrics(fObjectToWrite.getTupleDataSize(), true);
1557-
} catch (Exception e) {
1558-
String errRootCause = com.ibm.streamsx.objectstorage.Utils.getErrorRootCause(e);
1559-
String errMsg = Messages.getString("OBJECTSTORAGE_SINK_OBJECT_WRITE_FAILURE", getBucketName(), fObjectToWrite.getPath(), errRootCause);
1560-
if (TRACE.isLoggable(TraceLevel.ERROR)) {
1599+
if (fObjectToWrite != null) {
1600+
try {
1601+
fObjectToWrite.writeTuple(tuple, partitionKey, fOSObjectRegistry);
1602+
// update metrics
1603+
updateCachedDataMetrics(fObjectToWrite.getTupleDataSize(), true);
1604+
} catch (Exception e) {
1605+
String errRootCause = com.ibm.streamsx.objectstorage.Utils.getErrorRootCause(e);
1606+
String errMsg = Messages.getString("OBJECTSTORAGE_SINK_OBJECT_WRITE_FAILURE", getBucketName(), fObjectToWrite.getPath(), errRootCause);
15611607
TRACE.log(TraceLevel.ERROR, errMsg);
1562-
TRACE.log(TraceLevel.ERROR, "Failed to write to object '"
1563-
+ fObjectToWrite.getPath() + "' to bucket '" + getBucketName() + "'. Exception: " + e.getMessage());
1608+
TRACE.log(TraceLevel.ERROR, "Failed to write to object '" + fObjectToWrite.getPath() + "' to bucket '" + getBucketName() + "'. Exception: " + e.getMessage());
1609+
LOGGER.log(TraceLevel.ERROR, errMsg);
1610+
getWriteFailuresMetric().increment();
1611+
if (isConsistentRegion()) {
1612+
throw new Exception(e);
1613+
}
15641614
}
1565-
LOGGER.log(TraceLevel.ERROR, errMsg);
1566-
throw new Exception(e);
1615+
}
1616+
else {
1617+
TRACE.log(TraceLevel.ERROR, "No writable object");
15671618
}
15681619
}
15691620

‎com.ibm.streamsx.objectstorage/info.xml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,15 @@ Note, that other operator parameter groups (except of authentication and connect
2424

2525
This is an overview of changes for major and minor version upgrades. For details, see the releases in public [https://github.com/IBMStreams/streamsx.objectstorage/releases|GitHub].
2626

27+
++ What is changed in version 2.0.2
28+
29+
* Improved error handling when using ObjectStorageSink with closeOnPunct and output port. In case of connection/close failures the exception is not forwarded and the metric `nCloseFailures` is incremented.
30+
* Metric `nWriteFailures` added for number of failures during object create or writing to an object.
31+
2732
++ What is changed in version 2.0.1
2833

2934
* Issue resolved for ObjectStorageSink operator when running in consistent region and no output stream is connected. Failure during object close causes reset of the region now.
35+
* Metric `nCloseFailures` added for number of failures during object flush/close.
3036

3137
++ What is new in version 2.0.0
3238

@@ -187,7 +193,7 @@ When writing objects in *partitioned parquet* format both clients work similar a
187193
* stocator (cos protocol): buffers on local disk prior upload
188194

189195
</info:description>
190-
<info:version>2.0.1</info:version>
196+
<info:version>2.0.2</info:version>
191197
<info:requiredProductVersion>4.2.0.0</info:requiredProductVersion>
192198
</info:identity>
193199
<info:dependencies/>

‎com.ibm.streamsx.objectstorage/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
<groupId>com.ibm.streamsx.objectstorage</groupId>
1313
<artifactId>streamsx.objectstorage</artifactId>
1414
<packaging>jar</packaging>
15-
<version>2.0.1</version>
15+
<version>2.0.2</version>
1616
<name>com.ibm.streamsx.objectstorage</name>
1717
<repositories>
1818
<repository>

‎test/java/com.ibm.streamsx.objectstorage.test/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,7 @@
1212
<groupId>com.ibm.streamsx.objectstorage.test</groupId>
1313
<artifactId>streamsx.objectstorage</artifactId>
1414
<packaging>jar</packaging>
15-
<version>2.0.1</version>
15+
<version>2.0.2</version>
1616
<name>com.ibm.streamsx.objectstorage.test</name>
1717
<repositories>
1818
<repository>

‎test/performance/com.ibm.streamsx.objectstorage.s3.test/Makefile‎

Lines changed: 16 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,14 @@ SPL_CMD_ARGS ?=
3636

3737
SPL_MAIN_COMPOSITE0 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_consistent_region_HMAC
3838
SPL_MAIN_COMPOSITE1 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_sync_consistent_region_HMAC
39+
SPL_MAIN_COMPOSITE2 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_HMAC
40+
SPL_MAIN_COMPOSITE3 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_sync_HMAC
41+
SPL_MAIN_COMPOSITE4 = com.ibm.streamsx.objectstorage.s3.test::WriteParquet_HMAC
42+
SPL_MAIN_COMPOSITE5 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_consistent_region_HMAC_s3a
43+
SPL_MAIN_COMPOSITE6 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_sync_consistent_region_HMAC_s3a
44+
SPL_MAIN_COMPOSITE7 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_HMAC_s3a
45+
SPL_MAIN_COMPOSITE8 = com.ibm.streamsx.objectstorage.s3.test::WriteRaw_sync_HMAC_s3a
46+
SPL_MAIN_COMPOSITE9 = com.ibm.streamsx.objectstorage.s3.test::WriteParquet_HMAC_s3a
3947
#SPL_MAIN_COMPOSITE8 = com.ibm.streamsx.objectstorage.s3.test::WriteDurationTest
4048
#SPL_MAIN_COMPOSITE2 = com.ibm.streamsx.objectstorage.s3.test::WriteDurationTestIAM
4149
#SPL_MAIN_COMPOSITE3 = com.ibm.streamsx.objectstorage.s3.test::WriteDurationTestJava
@@ -64,12 +72,14 @@ $(BIN_LOC)/com/ibm/streamsx/objectstorage/perf/test/TestSource.class: impl/java/
6472
distributed: javacompile
6573
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE0) --output-directory=$(OUTPUT_DIR)/0 $(SPL_CMD_ARGS)
6674
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE1) --output-directory=$(OUTPUT_DIR)/1 $(SPL_CMD_ARGS)
67-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE2) --output-directory=$(OUTPUT_DIR)/2 $(SPL_CMD_ARGS)
68-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE3) --output-directory=$(OUTPUT_DIR)/3 $(SPL_CMD_ARGS)
69-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE4) --output-directory=$(OUTPUT_DIR)/4 $(SPL_CMD_ARGS)
70-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE5) --output-directory=$(OUTPUT_DIR)/5 $(SPL_CMD_ARGS)
71-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE6) --output-directory=$(OUTPUT_DIR)/6 $(SPL_CMD_ARGS)
72-
# $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE7) --output-directory=$(OUTPUT_DIR)/7 $(SPL_CMD_ARGS)
75+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE2) --output-directory=$(OUTPUT_DIR)/2 $(SPL_CMD_ARGS)
76+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE3) --output-directory=$(OUTPUT_DIR)/3 $(SPL_CMD_ARGS)
77+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE4) --output-directory=$(OUTPUT_DIR)/4 $(SPL_CMD_ARGS)
78+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE5) --output-directory=$(OUTPUT_DIR)/5 $(SPL_CMD_ARGS)
79+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE6) --output-directory=$(OUTPUT_DIR)/6 $(SPL_CMD_ARGS)
80+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE7) --output-directory=$(OUTPUT_DIR)/7 $(SPL_CMD_ARGS)
81+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE8) --output-directory=$(OUTPUT_DIR)/8 $(SPL_CMD_ARGS)
82+
$(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE9) --output-directory=$(OUTPUT_DIR)/9 $(SPL_CMD_ARGS)
7383

7484

7585
clean:

‎test/performance/com.ibm.streamsx.objectstorage.s3.test/com.ibm.streamsx.objectstorage.s3.test/WriteParquetHMAC_cos.spl‎

Lines changed: 117 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,117 @@
1+
//
2+
// ****************************************************************************
3+
// * Copyright (C) 2018, International Business Machines Corporation *
4+
// * All rights reserved. *
5+
// ****************************************************************************
6+
//
7+
8+
namespace com.ibm.streamsx.objectstorage.s3.test;
9+
10+
use com.ibm.streamsx.objectstorage.s3::*;
11+
use com.ibm.streamsx.objectstorage::*;
12+
13+
/**
14+
* This application tests to write objects to Object Storage with parquet format.
15+
*/
16+
public composite WriteParquet_HMAC {
17+
18+
param
19+
expression<rstring> $accessKeyID : getSubmissionTimeValue("os-access-key-id");
20+
expression<rstring> $secretAccessKey : getSubmissionTimeValue("os-secret-access-key");
21+
expression<rstring> $endpoint : getSubmissionTimeValue("os-endpoint", "s3.us.cloud-object-storage.appdomain.cloud");
22+
expression<rstring> $bucket : getSubmissionTimeValue("os-bucket");
23+
expression<int32> $uploadWorkersNum: (int32)getSubmissionTimeValue("uploadWorkersNum", "10");
24+
25+
graph
26+
27+
stream<rstring result> CRTest = WriteParquet_HMACComp() {
28+
param
29+
accessKeyID: $accessKeyID;
30+
secretAccessKey: $secretAccessKey;
31+
bucket: $bucket;
32+
endpoint: $endpoint;
33+
uploadWorkersNum: $uploadWorkersNum;
34+
}
35+
36+
}
37+
38+
@threading(model=manual)
39+
public composite WriteParquet_HMACComp (output DummyResult)
40+
{
41+
param
42+
expression<rstring> $accessKeyID;
43+
expression<rstring> $secretAccessKey;
44+
expression<rstring> $bucket;
45+
expression<rstring> $endpoint: "s3.us.cloud-object-storage.appdomain.cloud";
46+
expression<int32> $uploadWorkersNum: 10;
47+
expression<float64> $drainPeriod: 30.0;
48+
expression<int32> $maxActiveBlocks: 10;
49+
expression<int32> $multipartSize: 10485760;
50+
51+
type DataHistorianData_t =
52+
rstring id,
53+
rstring tz,
54+
rstring dateutc,
55+
float64 latitude,
56+
float64 longitude,
57+
float64 temperature,
58+
float64 baromin,
59+
float64 humidity,
60+
float64 rainin,
61+
rstring time_stamp,
62+
rstring fillData;
63+
64+
graph
65+
66+
stream<DataHistorianData_t> DataStream = Beacon() {
67+
output
68+
DataStream:
69+
id = "I53700CO" + (rstring)(random()*100.0),
70+
tz = "Europe/Paris",
71+
dateutc = createTimestamp(),
72+
latitude = random() * 100.0,
73+
longitude = random() * 100.0,
74+
temperature = random() * 50.0,
75+
baromin = random() * 40.0,
76+
humidity = random() * 70.0,
77+
rainin = random() * 20.0,
78+
time_stamp = createTimestamp(),
79+
fillData = getRandomData();
80+
config placement: partitionColocation("PERFTEST");
81+
}
82+
83+
() as OSink = S3ObjectStorageSink(DataStream) {
84+
param
85+
accessKeyID : $accessKeyID;
86+
secretAccessKey : $secretAccessKey;
87+
endpoint : $endpoint;
88+
bucket : $bucket;
89+
protocol: cos;
90+
sslEnabled: false;
91+
objectName: "test_data_%OBJECTNUM.snappy.parquet";
92+
storageFormat: "parquet";
93+
parquetCompression: "SNAPPY";
94+
parquetEnableDict : true;
95+
uploadWorkersNum: $uploadWorkersNum;
96+
s3aFastUploadActiveBlocks: $maxActiveBlocks;
97+
s3aFastUploadBuffer: "bytebuffer";
98+
s3aMultipartSize: $multipartSize;
99+
//parquetBlockSize: 33554432;
100+
//parquetPageSize: 2048;
101+
vmArg: "-Xmx 4096m";
102+
tuplesPerObject: 50000l;
103+
config placement: partitionColocation("PERFTEST");
104+
}
105+
106+
stream<rstring result> DummyResult = Custom() {
107+
logic
108+
onProcess: {
109+
while (true) {
110+
submit ({result=""}, DummyResult);
111+
block(1.0);
112+
}
113+
}
114+
}
115+
}
116+
117+

0 commit comments

Comments
 (0)