Skip to content

Commit 4cf2504

Browse files
dormstddormstd
committed
Fix: Changed the httpclient to be closeable to fix a hang when trying to backup dataset with multiple files
Signed-off-by: Ricardo Ruiz Saiz <dorm.std@gmail.com>
1 parent ca2c743 commit 4cf2504

2 files changed

Lines changed: 63 additions & 22 deletions

File tree

‎CKAN_API_Handler/src/main/java/net/atos/qrowd/handlers/CKAN_API_Handler.java‎

Lines changed: 58 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -20,15 +20,15 @@
2020
import net.atos.qrowd.pojos.*;
2121
import org.apache.commons.io.FileUtils;
2222
import org.apache.http.HttpEntity;
23-
import org.apache.http.HttpResponse;
23+
import org.apache.http.client.methods.CloseableHttpResponse;
2424
import org.apache.http.client.methods.HttpPost;
2525
import org.apache.http.entity.ContentType;
2626
import org.apache.http.entity.mime.MultipartEntityBuilder;
2727
import org.apache.http.entity.mime.content.ContentBody;
2828
import org.apache.http.entity.mime.content.FileBody;
2929
import org.apache.http.entity.mime.content.StringBody;
3030
import org.apache.http.impl.client.CloseableHttpClient;
31-
import org.apache.http.impl.client.HttpClientBuilder;
31+
import org.apache.http.impl.client.HttpClients;
3232
import org.apache.log4j.Logger;
3333

3434
import java.io.BufferedReader;
@@ -58,15 +58,15 @@ public CKAN_API_Handler(String HOST, String api_key, String filename, String org
5858
this.organization_id = organization_id.toLowerCase();
5959
this.package_private = package_private;
6060

61-
this.httpclient = HttpClientBuilder.create().build();
61+
this.httpclient = HttpClients.createDefault();
6262
}
6363

6464
public CKAN_API_Handler(String HOST, String api_key)
6565
{
6666
this.HOST = HOST;
6767
this.api_key = api_key;
6868

69-
this.httpclient = HttpClientBuilder.create().build();
69+
this.httpclient = HttpClients.createDefault();
7070
}
7171

7272
public boolean packageExists(String package_id) throws IOException{
@@ -79,13 +79,16 @@ public boolean packageExists(String package_id) throws IOException{
7979
postRequest = new HttpPost(HOST+"/api/3/action/package_search?q=name:"+package_id);
8080
postRequest.setHeader("X-CKAN-API-Key", api_key);
8181

82-
HttpResponse response = httpclient.execute(postRequest);
82+
CloseableHttpClient httpclient = HttpClients.createDefault();
83+
CloseableHttpResponse response = httpclient.execute(postRequest);
8384
int statusCode = response.getStatusLine().getStatusCode();
8485
BufferedReader br = new BufferedReader(
8586
new InputStreamReader((response.getEntity().getContent())));
8687
while ((line = br.readLine()) != null) {
8788
sb.append(line);
8889
}
90+
httpclient.close();
91+
response.close();
8992
// Parse the response into a POJO to be able to get results from it.
9093
// ToDo: If no result is returned, raise an error (when converting to POJO fails or return code !=200?)
9194
if(statusCode==200) {
@@ -115,13 +118,16 @@ public Package_ getPackageByName(String name) throws IOException {
115118
postRequest = new HttpPost(HOST+"/api/3/action/package_search?q=name:"+name);
116119
postRequest.setHeader("X-CKAN-API-Key", api_key);
117120

118-
HttpResponse response = httpclient.execute(postRequest);
121+
CloseableHttpClient httpclient = HttpClients.createDefault();
122+
CloseableHttpResponse response = httpclient.execute(postRequest);
119123
int statusCode = response.getStatusLine().getStatusCode();
120124
BufferedReader br = new BufferedReader(
121125
new InputStreamReader((response.getEntity().getContent())));
122126
while ((line = br.readLine()) != null) {
123127
sb.append(line);
124128
}
129+
httpclient.close();
130+
response.close();
125131
// Parse the response into a POJO to be able to get results from it.
126132
// ToDo: If no result is returned, raise an error (when converting to POJO fails or return code !=200?)
127133
if(statusCode==200) {
@@ -158,7 +164,8 @@ public void createPackage(String package_id) throws IOException{
158164
postRequest.setEntity(reqEntity);
159165
postRequest.setHeader("X-CKAN-API-Key", api_key);
160166

161-
HttpResponse response = httpclient.execute(postRequest);
167+
CloseableHttpClient httpclient = HttpClients.createDefault();
168+
CloseableHttpResponse response = httpclient.execute(postRequest);
162169
int statusCode = response.getStatusLine().getStatusCode();
163170

164171
BufferedReader br = new BufferedReader(
@@ -169,6 +176,9 @@ public void createPackage(String package_id) throws IOException{
169176
sb.append(line);
170177
sb.append("\n");
171178
}
179+
httpclient.close();
180+
response.close();
181+
172182
if(statusCode!=200){
173183
log.error("statusCode =!=" +statusCode);
174184
log.error("Error creating the package via CKAN API. Package id: "+package_id);
@@ -222,7 +232,8 @@ public void createPackagePojo(Package_ dataset, String name) throws IOException{
222232
postRequest.setEntity(reqEntity);
223233
postRequest.setHeader("X-CKAN-API-Key", api_key);
224234

225-
HttpResponse response = httpclient.execute(postRequest);
235+
CloseableHttpClient httpclient = HttpClients.createDefault();
236+
CloseableHttpResponse response = httpclient.execute(postRequest);
226237
int statusCode = response.getStatusLine().getStatusCode();
227238

228239
BufferedReader br = new BufferedReader(
@@ -233,6 +244,9 @@ public void createPackagePojo(Package_ dataset, String name) throws IOException{
233244
sb.append(line);
234245
sb.append("\n");
235246
}
247+
httpclient.close();
248+
response.close();
249+
236250
if(statusCode!=200){
237251
log.error("statusCode =!=" +statusCode);
238252
log.error(sb);
@@ -256,7 +270,8 @@ public boolean organizationExists() throws IOException{
256270
postRequest.setEntity(reqEntity);
257271
postRequest.setHeader("X-CKAN-API-Key", api_key);
258272

259-
HttpResponse response = httpclient.execute(postRequest);
273+
CloseableHttpClient httpclient = HttpClients.createDefault();
274+
CloseableHttpResponse response = httpclient.execute(postRequest);
260275
int statusCode = response.getStatusLine().getStatusCode();
261276

262277
BufferedReader br = new BufferedReader(
@@ -265,6 +280,8 @@ public boolean organizationExists() throws IOException{
265280
sb.append(line);
266281
sb.append("\n");
267282
}
283+
httpclient.close();
284+
response.close();
268285

269286
if(statusCode==200)
270287
{
@@ -294,7 +311,8 @@ public void createOrganization() throws IOException{
294311
postRequest.setEntity(reqEntity);
295312
postRequest.setHeader("X-CKAN-API-Key", api_key);
296313

297-
HttpResponse response = httpclient.execute(postRequest);
314+
CloseableHttpClient httpclient = HttpClients.createDefault();
315+
CloseableHttpResponse response = httpclient.execute(postRequest);
298316
int statusCode = response.getStatusLine().getStatusCode();
299317

300318
BufferedReader br = new BufferedReader(
@@ -305,6 +323,9 @@ public void createOrganization() throws IOException{
305323
sb.append(line);
306324
sb.append("\n");
307325
}
326+
httpclient.close();
327+
response.close();
328+
308329
if (statusCode != 200) {
309330
log.error("statusCode =!=" + statusCode);
310331
log.error(sb);
@@ -323,17 +344,15 @@ public void uploadFilePojo(Resource resource, String dataset_name, String resour
323344
file.deleteOnExit();
324345
FileUtils.copyURLToFile(url, file);
325346

326-
327-
HttpPost postRequest;
328347
ContentBody cbFile = new FileBody(file, ContentType.TEXT_HTML);
348+
HttpPost postRequest;
329349

330350
MultipartEntityBuilder multipart = MultipartEntityBuilder.create()
331351
.addPart("file", cbFile)
332352
.addPart("key", new StringBody(resourceFileName.split("\\.")[0],ContentType.TEXT_PLAIN))
333353
.addPart("name", new StringBody(resourceFileName,ContentType.TEXT_PLAIN))
334354
.addPart("package_id",new StringBody(dataset_name,ContentType.TEXT_PLAIN))
335355
.addPart("upload",cbFile);
336-
//ToDo: Add the rest of the fields¿?
337356
if(resource.getUrl() != null){
338357
multipart.addPart("url",new StringBody(resource.getUrl(),ContentType.TEXT_PLAIN));
339358
}
@@ -350,17 +369,28 @@ public void uploadFilePojo(Resource resource, String dataset_name, String resour
350369
}
351370
HttpEntity reqEntity = multipart.build();
352371

353-
postRequest = new HttpPost(HOST+"/api/action/resource_create");
372+
postRequest = new HttpPost(HOST+"/api/3/action/resource_create");
354373
postRequest.setEntity(reqEntity);
355374
postRequest.setHeader("X-CKAN-API-Key", api_key);
356375

357-
HttpResponse response = httpclient.execute(postRequest);
376+
CloseableHttpClient httpclient = HttpClients.createDefault();
377+
CloseableHttpResponse response = httpclient.execute(postRequest);
358378
int statusCode = response.getStatusLine().getStatusCode();
359379

380+
String line;
381+
StringBuilder sb = new StringBuilder();
382+
BufferedReader br = new BufferedReader(
383+
new InputStreamReader((response.getEntity().getContent())));
384+
while ((line = br.readLine()) != null) {
385+
sb.append(line);
386+
}
360387
if(statusCode!=200){
361388
log.error("statusCode =!=" +statusCode);
389+
log.error(sb.toString());
362390
}
363391
else log.info("Request returns statusCode 200: OK");
392+
response.close();
393+
httpclient.close();
364394
}
365395

366396
public Boolean createOrUpdateResource(String path) throws IOException {
@@ -376,12 +406,16 @@ public Boolean createOrUpdateResource(String path) throws IOException {
376406
postRequest = new HttpPost(HOST+"/api/action/resource_search?query=name:"+filename);
377407
postRequest.setHeader("X-CKAN-API-Key", api_key);
378408

379-
HttpResponse response = httpclient.execute(postRequest);
409+
CloseableHttpClient httpclient = HttpClients.createDefault();
410+
CloseableHttpResponse response = httpclient.execute(postRequest);
380411
BufferedReader br = new BufferedReader(
381412
new InputStreamReader((response.getEntity().getContent())));
382413
while ((line = br.readLine()) != null) {
383414
sb.append(line);
384415
}
416+
httpclient.close();
417+
response.close();
418+
385419
//Parse the response into a POJO to be able to get results from it.
386420
ResourceResponse resResponse = gson.fromJson(sb.toString(),ResourceResponse.class);
387421
System.out.println(resResponse);
@@ -469,8 +503,11 @@ private void updateFile(String path, String resourceId) throws IOException {
469503
postRequest.setEntity(reqEntity);
470504
postRequest.setHeader("X-CKAN-API-Key", api_key);
471505

472-
HttpResponse response = httpclient.execute(postRequest);
506+
CloseableHttpClient httpclient = HttpClients.createDefault();
507+
CloseableHttpResponse response = httpclient.execute(postRequest);
473508
int statusCode = response.getStatusLine().getStatusCode();
509+
httpclient.close();
510+
response.close();
474511

475512
if(statusCode!=200){
476513
log.error("statusCode =!=" +statusCode);
@@ -505,7 +542,8 @@ private void uploadFile(String path) throws IOException {
505542
postRequest.setEntity(reqEntity);
506543
postRequest.setHeader("X-CKAN-API-Key", api_key);
507544

508-
HttpResponse response = httpclient.execute(postRequest);
545+
CloseableHttpClient httpclient = HttpClients.createDefault();
546+
CloseableHttpResponse response = httpclient.execute(postRequest);
509547
int statusCode = response.getStatusLine().getStatusCode();
510548

511549
BufferedReader br = new BufferedReader(
@@ -516,6 +554,8 @@ private void uploadFile(String path) throws IOException {
516554
sb.append(line);
517555
sb.append("\n");
518556
}
557+
httpclient.close();
558+
response.close();
519559

520560
if(statusCode!=200){
521561
log.error("Error creating a resource: "+ file.getName().split("\\.")[0] +"in package:"+package_id);
@@ -527,7 +567,6 @@ private void uploadFile(String path) throws IOException {
527567

528568
public void close()
529569
{
530-
//httpclient.getConnectionManager().shutdown();
531570
try {
532571
httpclient.close();
533572
} catch (IOException e) {

‎nifi-nifiCKANDatasetBackup-processors/src/main/java/net/atos/qrowd/processors/nifiCKANDatasetBackup/CKAN_Package_Backup.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -169,7 +169,7 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
169169
ckan_api_handler.uploadFilePojo(res, datasetName, resourceFileName);
170170
}
171171
}
172-
//Transfer the input file through succes relationship
172+
//Transfer the input file through success relationship
173173
session.transfer(flowFile, REL_BACKUP_CREATED);
174174
ckan_api_handler.close();
175175
}else
@@ -187,8 +187,10 @@ public void onTrigger(final ProcessContext context, final ProcessSession session
187187
getLogger().log(LogLevel.ERROR, "Error while splitting the resource filename, it contains no '.'");
188188
getLogger().error(oob.toString());
189189
session.transfer(session.penalize(flowFile), REL_FAILURE);
190+
}finally {
191+
// As long as we commit the session right here, we are safe.
192+
session.commit();
193+
getLogger().info("Processor finished completely");
190194
}
191-
// As long as we commit the session right here, we are safe.
192-
session.commit();
193195
}
194196
}

0 commit comments

Comments
 (0)