Skip to content

Commit c7ef6c7

Browse files
author
psainics
committed
Add support for HTTP PATCH
1 parent b0019d4 commit c7ef6c7

7 files changed

Lines changed: 62 additions & 13 deletions

File tree

‎src/main/java/io/cdap/plugin/http/sink/batch/HTTPRecordWriter.java‎

Lines changed: 33 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -22,9 +22,11 @@
2222
import io.cdap.cdap.api.data.format.StructuredRecord;
2323
import io.cdap.cdap.api.data.schema.Schema;
2424
import io.cdap.plugin.http.common.RetryPolicy;
25+
import io.cdap.plugin.http.common.error.ErrorHandling;
2526
import io.cdap.plugin.http.common.error.HttpErrorHandler;
2627
import io.cdap.plugin.http.common.error.RetryableErrorHandling;
2728
import io.cdap.plugin.http.common.http.HttpRequest;
29+
import io.cdap.plugin.http.common.http.HttpResponse;
2830
import io.cdap.plugin.http.common.http.OAuthUtil;
2931

3032
import org.apache.hadoop.mapreduce.RecordWriter;
@@ -83,6 +85,7 @@ public class HTTPRecordWriter extends RecordWriter<StructuredRecord, StructuredR
8385
public static final String REQUEST_METHOD_POST = "POST";
8486
public static final String REQUEST_METHOD_PUT = "PUT";
8587
public static final String REQUEST_METHOD_DELETE = "DELETE";
88+
public static final String REQUEST_METHOD_PATCH = "PATCH";
8689

8790
private final HTTPSinkConfig config;
8891
private final MessageBuffer messageBuffer;
@@ -96,6 +99,7 @@ public class HTTPRecordWriter extends RecordWriter<StructuredRecord, StructuredR
9699
private final HttpErrorHandler httpErrorHandler;
97100
private final PollInterval pollInterval;
98101
private int httpStatusCode;
102+
private String httpResponseBody;
99103
private static int retryCount;
100104

101105
HTTPRecordWriter(HTTPSinkConfig config, Schema inputSchema) {
@@ -120,11 +124,13 @@ public class HTTPRecordWriter extends RecordWriter<StructuredRecord, StructuredR
120124
@Override
121125
public void write(StructuredRecord input, StructuredRecord unused) throws IOException {
122126
configURL = url;
123-
if (config.getMethod().equals(REQUEST_METHOD_POST) || config.getMethod().equals(REQUEST_METHOD_PUT)) {
127+
if (config.getMethod().equals(REQUEST_METHOD_POST) || config.getMethod().equals(REQUEST_METHOD_PUT) ||
128+
config.getMethod().equals(REQUEST_METHOD_PATCH)) {
124129
messageBuffer.add(input);
125130
}
126131

127-
if (config.getMethod().equals(REQUEST_METHOD_PUT) || config.getMethod().equals(REQUEST_METHOD_DELETE)
132+
if (config.getMethod().equals(REQUEST_METHOD_PUT) || config.getMethod().equals(REQUEST_METHOD_PATCH) ||
133+
config.getMethod().equals(REQUEST_METHOD_DELETE)
128134
&& !placeHolderList.isEmpty()) {
129135
configURL = updateURLWithPlaceholderValue(input);
130136
}
@@ -200,9 +206,9 @@ private boolean executeHTTPServiceAndCheckStatusCode() throws IOException {
200206
request.setHeaders(getRequestHeaders());
201207

202208
response = httpClient.execute(request);
203-
204209
httpStatusCode = response.getStatusLine().getStatusCode();
205210
LOG.debug("Response HTTP Status code: {}", httpStatusCode);
211+
httpResponseBody = new HttpResponse(response).getBody();
206212

207213
} catch (MalformedURLException | ProtocolException e) {
208214
throw new IllegalStateException("Error opening url connection. Reason: " + e.getMessage(), e);
@@ -277,7 +283,9 @@ private Header[] getRequestHeaders() throws IOException {
277283
headers.put("Instance-Follow-Redirects", String.valueOf(config.getFollowRedirects()));
278284
headers.put("charset", config.getCharset());
279285

280-
if (config.getMethod().equals(REQUEST_METHOD_POST) || config.getMethod().equals(REQUEST_METHOD_PUT)) {
286+
if (config.getMethod().equals(REQUEST_METHOD_POST)
287+
|| config.getMethod().equals(REQUEST_METHOD_PATCH)
288+
|| config.getMethod().equals(REQUEST_METHOD_PUT)) {
281289
if (!headers.containsKey("Content-Type")) {
282290
headers.put("Content-Type", contentType);
283291
}
@@ -302,7 +310,8 @@ private Header getAuthorizationHeader(AccessToken accessToken) {
302310
*/
303311
private List<PlaceholderBean> getPlaceholderListFromURL() {
304312
List<PlaceholderBean> placeholderList = new ArrayList<>();
305-
if (!(config.getMethod().equals(REQUEST_METHOD_PUT) || config.getMethod().equals(REQUEST_METHOD_DELETE))) {
313+
if (!(config.getMethod().equals(REQUEST_METHOD_PUT) || config.getMethod().equals(REQUEST_METHOD_PATCH) ||
314+
config.getMethod().equals(REQUEST_METHOD_DELETE))) {
306315
return placeholderList;
307316
}
308317
Pattern pattern = Pattern.compile(REGEX_HASHED_VAR);
@@ -351,6 +360,25 @@ private void flushMessageBuffer() {
351360
"after the batch execution. " + e);
352361
}
353362
messageBuffer.clear();
363+
364+
ErrorHandling postRetryStrategy = httpErrorHandler.getErrorHandlingStrategy(httpStatusCode)
365+
.getAfterRetryStrategy();
366+
367+
switch (postRetryStrategy) {
368+
case SUCCESS:
369+
break;
370+
case STOP:
371+
throw new IllegalStateException(String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
372+
config.getUrl(), httpStatusCode, httpResponseBody));
373+
case SKIP:
374+
case SEND:
375+
LOG.warn(String.format("Fetching from url '%s' returned status code '%d' and body '%s'",
376+
config.getUrl(), httpStatusCode, httpResponseBody));
377+
break;
378+
default:
379+
throw new IllegalArgumentException(String.format("Unexpected http error handling: '%s'", postRetryStrategy));
380+
}
381+
354382
}
355383

356384
}

‎src/main/java/io/cdap/plugin/http/sink/batch/HTTPSinkConfig.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -80,7 +80,7 @@ public class HTTPSinkConfig extends BaseHttpConfig {
8080
private static final String REGEX_HASHED_VAR = "#(\\w+)";
8181
private static final String PLACEHOLDER = "#";
8282
private static final Set<String> METHODS = ImmutableSet.of(HttpMethod.GET, HttpMethod.POST,
83-
HttpMethod.PUT, HttpMethod.DELETE);
83+
HttpMethod.PUT, HttpMethod.DELETE, "PATCH");
8484

8585
@Name(URL)
8686
@Description("The URL to post data to. Additionally, a placeholder like #columnName can be added to the URL that " +
@@ -462,7 +462,6 @@ public void validate(FailureCollector collector) {
462462
if (!containsMacro(PROPERTY_RETRY_POLICY) && getRetryPolicy() == RetryPolicy.LINEAR) {
463463
assertIsSet(getLinearRetryInterval(), PROPERTY_LINEAR_RETRY_INTERVAL, "retry policy is linear");
464464
}
465-
466465
if (!containsMacro(READ_TIMEOUT) && Objects.nonNull(readTimeout) && readTimeout < 0) {
467466
collector.addFailure("Read Timeout cannot be a negative number.", null)
468467
.withConfigProperty(READ_TIMEOUT);
@@ -494,7 +493,7 @@ public void validateSchema(@Nullable Schema schema, FailureCollector collector)
494493
return;
495494
}
496495

497-
if ((method.equals("PUT") || method.equals("DELETE")) && url.contains(PLACEHOLDER)) {
496+
if ((method.equals("PUT") || method.equals("PATCH") || method.equals("DELETE")) && url.contains(PLACEHOLDER)) {
498497
Pattern pattern = Pattern.compile(REGEX_HASHED_VAR);
499498
Matcher matcher = pattern.matcher(url);
500499
List<String> fieldNames = fields.stream().map(field -> field.getName()).collect(Collectors.toList());

‎src/main/java/io/cdap/plugin/http/source/batch/HttpInputFormatProvider.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import io.cdap.plugin.http.common.pagination.page.PageFormat;
2929
import io.cdap.plugin.http.source.common.DelimitedSchemaDetector;
3030
import io.cdap.plugin.http.source.common.RawStringPerLine;
31+
import org.apache.http.client.methods.CloseableHttpResponse;
3132

3233
import java.io.IOException;
3334
import java.util.Map;
@@ -68,8 +69,14 @@ public Schema getSchema(FailureCollector failureCollector) {
6869
case TSV:
6970
String delimiter = format == PageFormat.CSV ? "," : "\t";
7071
try (HttpClient client = new HttpClient(config)) {
71-
RawStringPerLine rawStringPerLine = new RawStringPerLine(
72-
new HttpResponse(client.executeHTTP(config.getUrl())));
72+
CloseableHttpResponse closeableHttpResponse = client.executeHTTP(config.getUrl());
73+
int statusCode = closeableHttpResponse.getStatusLine().getStatusCode();
74+
if (statusCode < 200 || statusCode >= 300) {
75+
failureCollector.addFailure(String.format("Failed to read the file, non 2xx status code received: %s",
76+
statusCode), null);
77+
return null;
78+
}
79+
RawStringPerLine rawStringPerLine = new RawStringPerLine(new HttpResponse(closeableHttpResponse));
7380
return DelimitedSchemaDetector.detectSchema(config, delimiter, rawStringPerLine, failureCollector);
7481
} catch (IOException e) {
7582
String errorMessage = e.getMessage();

‎src/test/java/io/cdap/plugin/http/sink/batch/HTTPSinkConfigTest.java‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -154,6 +154,18 @@ public void testValidInputSchema() {
154154
Assert.assertTrue(collector.getValidationFailures().isEmpty());
155155
}
156156

157+
@Test()
158+
public void testValidInputWithPlaceHoldersWithPATCH() {
159+
Schema schema = Schema.recordOf("record",
160+
Schema.Field.of("id", Schema.of(Schema.Type.LONG)),
161+
Schema.Field.of("name", Schema.of(Schema.Type.STRING)));
162+
String dynamicUrl = "http://example.com/api/v1/book/#id";
163+
HTTPSinkConfig config = HTTPSinkConfig.newBuilder(VALID_CONFIG).setMethod("PATCH").setUrl(dynamicUrl).build();
164+
MockFailureCollector collector = new MockFailureCollector("httpsinkwithvalidinputschema");
165+
config.validateSchema(schema, collector);
166+
Assert.assertTrue(collector.getValidationFailures().isEmpty());
167+
}
168+
157169
@Test(expected = ValidationException.class)
158170
public void testHTTPSinkWithNegativeBatchSize() {
159171
HTTPSinkConfig config = HTTPSinkConfig.newBuilder(VALID_CONFIG)

‎widgets/HTTP-batchsink.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,8 @@
2525
"POST",
2626
"GET",
2727
"PUT",
28-
"DELETE"
28+
"DELETE",
29+
"PATCH"
2930
],
3031
"default": "POST"
3132
}

‎widgets/HTTP-batchsource.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@
2626
"POST",
2727
"PUT",
2828
"DELETE",
29-
"HEAD"
29+
"HEAD",
30+
"PATCH"
3031
],
3132
"default": "GET"
3233
}

‎widgets/HTTP-streamingsource.json‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,7 +26,8 @@
2626
"POST",
2727
"PUT",
2828
"DELETE",
29-
"HEAD"
29+
"HEAD",
30+
"PATCH"
3031
],
3132
"default": "GET"
3233
}

0 commit comments

Comments
 (0)