Skip to content

Commit 0833dde

Browse files
stas-panasiukstas-panasiuk
committed
conflicts resolved
1 parent a22bb63 commit 0833dde

File tree

15 files changed

+312
-275
lines changed

15 files changed

+312
-275
lines changed

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/request/ClusterPipelineRequest.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,13 +2,12 @@
22

33
import com.fasterxml.jackson.annotation.JsonCreator;
44
import com.fasterxml.jackson.annotation.JsonProperty;
5+
import java.util.List;
56
import lombok.EqualsAndHashCode;
67
import lombok.Getter;
78
import lombok.Setter;
89
import lombok.ToString;
910

10-
import java.util.List;
11-
1211

1312
@Getter
1413
@Setter

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/request/ClusterRequest.java‎

Lines changed: 2 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,16 @@
11
package com.cloudera.service.common.request;
22

3+
import java.util.List;
34
import lombok.EqualsAndHashCode;
45
import lombok.Getter;
56
import lombok.Setter;
67
import lombok.ToString;
78

8-
import java.util.List;
9-
109
@Getter
1110
@Setter
1211
@EqualsAndHashCode(callSuper = true)
1312
@ToString
14-
public class ClusterRequest extends AbstractRequest{
13+
public class ClusterRequest extends AbstractRequest {
1514
private final String clusterId;
1615
private final List<String> jobs;
1716

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/request/RequestType.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package com.cloudera.service.common.request;
22

33
public enum RequestType {
4-
GET_ALL_CLUSTERS_SERVICE_REQUEST, GET_CLUSTER_SERVICE_REQUEST, START_JOB_REQUEST, RESTART_JOB_REQUEST, STOP_JOB_REQUEST, GET_JOB_CONFIG_REQUEST, CREATE_EMPTY_PIPELINE, START_PIPELINE, UPDATE_JOB_CONFIG_REQUEST
4+
GET_ALL_CLUSTERS_SERVICE_REQUEST, GET_CLUSTER_SERVICE_REQUEST, START_JOB_REQUEST, RESTART_JOB_REQUEST,
5+
STOP_JOB_REQUEST, GET_JOB_CONFIG_REQUEST, CREATE_EMPTY_PIPELINE, START_PIPELINE, UPDATE_JOB_CONFIG_REQUEST
56
}

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/response/Job.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -54,8 +54,9 @@ public String[] getScript(Job job) {
5454
case INDEX:
5555
case TRIAGE:
5656
return new String[]{scriptName, job.getJobBranch(), job.getJobPipeline()};
57+
default:
58+
return new String[]{};
5759
}
58-
return new String[]{};
5960
}
6061

6162
}

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/response/Pipeline.java‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,11 @@
11
package com.cloudera.service.common.response;
22

3+
import java.util.List;
34
import lombok.AllArgsConstructor;
45
import lombok.Builder;
56
import lombok.Data;
67
import lombok.NoArgsConstructor;
78

8-
import java.util.List;
9-
109
@Data
1110
@Builder
1211
@NoArgsConstructor

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/response/ResponseType.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,7 @@
11
package com.cloudera.service.common.response;
22

33
public enum ResponseType {
4-
GET_ALL_CLUSTERS_SERVICE_RESPONSE, GET_CLUSTER_SERVICE_RESPONSE, START_JOB_RESPONSE, RESTART_JOB_RESPONSE, STOP_JOB_RESPONSE, GET_JOB_CONFIG_RESPONSE, UPDATE_JOB_CONFIG_RESPONSE, CREATE_EMPTY_PIPELINE_RESPONSE, START_PIPELINE_RESPONSE, ERROR_RESPONSE
4+
GET_ALL_CLUSTERS_SERVICE_RESPONSE, GET_CLUSTER_SERVICE_RESPONSE, START_JOB_RESPONSE, RESTART_JOB_RESPONSE,
5+
STOP_JOB_RESPONSE, GET_JOB_CONFIG_RESPONSE, UPDATE_JOB_CONFIG_RESPONSE, CREATE_EMPTY_PIPELINE_RESPONSE,
6+
START_PIPELINE_RESPONSE, ERROR_RESPONSE
57
}

‎flink-cyber/cyber-services/cyber-service-common/src/main/java/com/cloudera/service/common/utils/ArchiveUtil.java‎

Lines changed: 32 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,5 @@
11
package com.cloudera.service.common.utils;
22

3-
import lombok.experimental.UtilityClass;
4-
import lombok.extern.slf4j.Slf4j;
5-
import org.apache.commons.compress.archivers.ArchiveEntry;
6-
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
7-
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
8-
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
9-
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
10-
import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
11-
import org.apache.commons.lang3.tuple.Pair;
12-
import org.apache.flink.core.fs.FileStatus;
13-
143
import java.io.BufferedInputStream;
154
import java.io.BufferedOutputStream;
165
import java.io.ByteArrayInputStream;
@@ -25,6 +14,16 @@
2514
import java.nio.file.StandardCopyOption;
2615
import java.util.Base64;
2716
import java.util.List;
17+
import lombok.experimental.UtilityClass;
18+
import lombok.extern.slf4j.Slf4j;
19+
import org.apache.commons.compress.archivers.ArchiveEntry;
20+
import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
21+
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
22+
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
23+
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
24+
import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
25+
import org.apache.commons.lang3.tuple.Pair;
26+
import org.apache.flink.core.fs.FileStatus;
2827

2928
@Slf4j
3029
@UtilityClass
@@ -59,19 +58,19 @@ public static byte[] compressToTarGzInMemory(List<Pair<String, byte[]>> files) {
5958
return bos.toByteArray();
6059
}
6160
try (BufferedOutputStream buffOut = new BufferedOutputStream(bos);
62-
GzipCompressorOutputStream gzOut = new GzipCompressorOutputStream(buffOut);
63-
TarArchiveOutputStream tOut = new TarArchiveOutputStream(gzOut)) {
61+
GzipCompressorOutputStream gzOut = new GzipCompressorOutputStream(buffOut);
62+
TarArchiveOutputStream tarOut = new TarArchiveOutputStream(gzOut)) {
6463
try {
6564
for (Pair<String, byte[]> file : files) {
6665
TarArchiveEntry tarEntry = new TarArchiveEntry(
67-
file.getLeft());
66+
file.getLeft());
6867
tarEntry.setSize(file.getRight().length);
69-
tOut.putArchiveEntry(tarEntry);
70-
tOut.write(file.getRight());
71-
tOut.closeArchiveEntry();
68+
tarOut.putArchiveEntry(tarEntry);
69+
tarOut.write(file.getRight());
70+
tarOut.closeArchiveEntry();
7271
}
7372
} finally {
74-
tOut.finish();
73+
tarOut.finish();
7574
}
7675
}
7776
return bos.toByteArray();
@@ -88,20 +87,21 @@ private static void compressToTarGz(String inputPath, OutputStream outputStream)
8887
}
8988

9089
try (BufferedOutputStream buffOut = new BufferedOutputStream(outputStream);
91-
GzipCompressorOutputStream gzOut = new GzipCompressorOutputStream(buffOut);
92-
TarArchiveOutputStream tOut = new TarArchiveOutputStream(gzOut)) {
90+
GzipCompressorOutputStream gzOut = new GzipCompressorOutputStream(buffOut);
91+
TarArchiveOutputStream tarOut = new TarArchiveOutputStream(gzOut)) {
9392

9493
try {
9594
for (FileStatus file : fileList) {
96-
addFileToTar(tOut, file, inputPath);
95+
addFileToTar(tarOut, file, inputPath);
9796
}
9897
} finally {
99-
tOut.finish();
98+
tarOut.finish();
10099
}
101100
}
102101
}
103102

104-
private static void addFileToTar(TarArchiveOutputStream tOut, FileStatus file, String rootPath) throws IOException {
103+
private static void addFileToTar(TarArchiveOutputStream tarOut, FileStatus file, String rootPath)
104+
throws IOException {
105105
final Path filePath = Paths.get(file.getPath().getPath());
106106
String pathInsideTar;
107107
if (filePath.startsWith(rootPath)) {
@@ -114,12 +114,12 @@ private static void addFileToTar(TarArchiveOutputStream tOut, FileStatus file, S
114114
}
115115

116116
TarArchiveEntry tarEntry = new TarArchiveEntry(
117-
filePath.toFile(),
118-
pathInsideTar);
117+
filePath.toFile(),
118+
pathInsideTar);
119119

120-
tOut.putArchiveEntry(tarEntry);
121-
Files.copy(filePath, tOut);
122-
tOut.closeArchiveEntry();
120+
tarOut.putArchiveEntry(tarEntry);
121+
Files.copy(filePath, tarOut);
122+
tarOut.closeArchiveEntry();
123123
}
124124

125125
public static void decompressFromTarGzFile(String pathToTar, String outputPath) throws IOException {
@@ -136,7 +136,8 @@ public static void decompressFromTarGzInMemory(byte[] rawData, String outputPath
136136
decompressFromTarGzInMemory(rawData, outputPath, false);
137137
}
138138

139-
public static void decompressFromTarGzInMemory(byte[] rawData, String outputPath, boolean base64) throws IOException {
139+
public static void decompressFromTarGzInMemory(byte[] rawData, String outputPath, boolean base64)
140+
throws IOException {
140141
if (rawData == null) {
141142
throw new IOException("Provided null as .tar.gz data which is not allowed!");
142143
}
@@ -154,8 +155,8 @@ public static void decompressFromTarGzInMemory(byte[] rawData, String outputPath
154155

155156
private static void decompressFromTarGz(InputStream inputStream, String outputPath) throws IOException {
156157
try (BufferedInputStream bi = new BufferedInputStream(inputStream);
157-
GzipCompressorInputStream gzi = new GzipCompressorInputStream(bi);
158-
TarArchiveInputStream ti = new TarArchiveInputStream(gzi)) {
158+
GzipCompressorInputStream gzi = new GzipCompressorInputStream(bi);
159+
TarArchiveInputStream ti = new TarArchiveInputStream(gzi)) {
159160

160161
ArchiveEntry entry;
161162
while ((entry = ti.getNextEntry()) != null) {

‎flink-cyber/cyber-services/cyber-worker-service/src/main/java/com/cloudera/cyber/restcli/controller/KafkaListenerController.java‎

Lines changed: 40 additions & 29 deletions
Original file line numberDiff line numberDiff line change
@@ -1,15 +1,18 @@
11
package com.cloudera.cyber.restcli.controller;
22

33
import com.cloudera.cyber.restcli.configuration.AppWorkerConfig;
4-
import com.cloudera.cyber.restcli.service.JobService;
54
import com.cloudera.cyber.restcli.service.FilePipelineService;
5+
import com.cloudera.cyber.restcli.service.JobService;
66
import com.cloudera.service.common.Utils;
77
import com.cloudera.service.common.request.RequestBody;
88
import com.cloudera.service.common.request.RequestType;
99
import com.cloudera.service.common.response.ClusterMeta;
1010
import com.cloudera.service.common.response.Job;
1111
import com.cloudera.service.common.response.ResponseBody;
1212
import com.cloudera.service.common.response.ResponseType;
13+
import java.io.IOException;
14+
import java.util.Collections;
15+
import java.util.List;
1316
import lombok.RequiredArgsConstructor;
1417
import lombok.extern.slf4j.Slf4j;
1518
import org.springframework.kafka.annotation.KafkaListener;
@@ -22,10 +25,6 @@
2225
import org.springframework.messaging.support.MessageHeaderAccessor;
2326
import org.springframework.stereotype.Component;
2427

25-
import java.io.IOException;
26-
import java.util.Collections;
27-
import java.util.List;
28-
2928
@Component
3029
@RequiredArgsConstructor
3130
@Slf4j
@@ -37,7 +36,9 @@ public class KafkaListenerController {
3736
//TODO: Rewrite to Spring events. Probably split the events into separate types, such as cluster event, job event, pipeline event, etc.
3837
@KafkaListener(topics = "#{kafkaProperties.getRequestTopic()}", containerFactory = "kafkaListenerContainerFactory")
3938
@SendTo({"#{kafkaProperties.getReplyTopic()}"})
40-
public Message<ResponseBody> handleMessage(RequestBody requestBody, @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key, @Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
39+
public Message<ResponseBody> handleMessage(RequestBody requestBody,
40+
@Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
41+
@Header(KafkaHeaders.REPLY_TOPIC) byte[] replyTo,
4142
@Header(KafkaHeaders.CORRELATION_ID) byte[] correlationId) {
4243
log.info("Start processing message\n Message key: '{}' \n value: '{}'", key, requestBody);
4344

@@ -53,18 +54,19 @@ public Message<ResponseBody> handleMessage(RequestBody requestBody, @Header(Kafk
5354
try {
5455
Job job = jobService.restartJob(requestBody.getJobIdHex());
5556
ResponseBody responseBody = ResponseBody.builder()
56-
.jobs(Collections.singletonList(job))
57-
.build();
58-
return buildResponseMessage(responseBody, ResponseType.RESTART_JOB_RESPONSE, replyTo, correlationId);
57+
.jobs(Collections.singletonList(job))
58+
.build();
59+
return buildResponseMessage(responseBody, ResponseType.RESTART_JOB_RESPONSE, replyTo,
60+
correlationId);
5961
} catch (IOException e) {
6062
return handleErrorResponse(e, replyTo, correlationId);
6163
}
6264
case STOP_JOB_REQUEST:
6365
try {
6466
Job job = jobService.stopJob(requestBody.getJobIdHex());
6567
ResponseBody responseBody = ResponseBody.builder()
66-
.jobs(Collections.singletonList(job))
67-
.build();
68+
.jobs(Collections.singletonList(job))
69+
.build();
6870
return buildResponseMessage(responseBody, ResponseType.STOP_JOB_RESPONSE, replyTo, correlationId);
6971
} catch (IOException e) {
7072
return handleErrorResponse(e, replyTo, correlationId);
@@ -75,46 +77,53 @@ public Message<ResponseBody> handleMessage(RequestBody requestBody, @Header(Kafk
7577
try {
7678
jobService.updateConfig(requestBody.getPayload());
7779
final ResponseBody responseBody = ResponseBody.builder().build();
78-
return buildResponseMessage(responseBody, ResponseType.UPDATE_JOB_CONFIG_RESPONSE, replyTo, correlationId);
80+
return buildResponseMessage(responseBody, ResponseType.UPDATE_JOB_CONFIG_RESPONSE, replyTo,
81+
correlationId);
7982
} catch (IOException e) {
8083
return handleErrorResponse(e, replyTo, correlationId);
8184
}
8285
case CREATE_EMPTY_PIPELINE:
8386
try {
8487
pipelineService.createEmptyPipeline(requestBody.getPipelineName(), requestBody.getBranch());
8588
final ResponseBody responseBody = ResponseBody.builder().build();
86-
return buildResponseMessage(responseBody, ResponseType.CREATE_EMPTY_PIPELINE_RESPONSE, replyTo, correlationId);
89+
return buildResponseMessage(responseBody, ResponseType.CREATE_EMPTY_PIPELINE_RESPONSE, replyTo,
90+
correlationId);
8791
} catch (Exception e) {
8892
return handleErrorResponse(e, replyTo, correlationId);
8993
}
9094
case START_PIPELINE:
9195
try {
92-
pipelineService.extractPipeline(requestBody.getPayload(), requestBody.getPipelineName(), requestBody.getBranch());
93-
pipelineService.startPipelineJob(requestBody.getPipelineName(), requestBody.getBranch(), requestBody.getProfileName(), requestBody.getProfileName(), requestBody.getJobs());
96+
pipelineService.extractPipeline(requestBody.getPayload(), requestBody.getPipelineName(),
97+
requestBody.getBranch());
98+
pipelineService.startPipelineJob(requestBody.getPipelineName(), requestBody.getBranch(),
99+
requestBody.getProfileName(), requestBody.getProfileName(), requestBody.getJobs());
94100
final ResponseBody responseBody = ResponseBody.builder().build();
95-
return buildResponseMessage(responseBody, ResponseType.START_PIPELINE_RESPONSE, replyTo, correlationId);
101+
return buildResponseMessage(responseBody, ResponseType.START_PIPELINE_RESPONSE, replyTo,
102+
correlationId);
96103
} catch (Exception e) {
97104
log.error("Exception while processing the Start All request {}", e.getMessage());
98105
return handleErrorResponse(e, replyTo, correlationId);
99106

100107
}
101-
108+
default:
109+
return null;
102110
}
103111
return null;
104112
}
105113

106-
private Message<ResponseBody> getResponseBodyMessage(byte[] replyTo, byte[] correlationId, ResponseType responseType) {
114+
private Message<ResponseBody> getResponseBodyMessage(byte[] replyTo, byte[] correlationId,
115+
ResponseType responseType) {
107116
try {
108117
List<Job> jobs = jobService.getJobs();
109118
ResponseBody responseBody = ResponseBody.builder()
110-
.jobs(jobs)
111-
.clusterMeta(ClusterMeta.builder()
112-
.name(config.getName())
113-
.clusterId(config.getId())
114-
.clusterStatus(config.getStatus())
115-
.version(config.getVersion())
116-
.build())
117-
.build();
119+
.jobs(jobs)
120+
.clusterMeta(ClusterMeta.builder()
121+
.name(config.getName())
122+
.clusterId(config.getId())
123+
.clusterStatus(config.getStatus())
124+
.version(config.getVersion())
125+
.build())
126+
.build();
118127
return buildResponseMessage(responseBody, responseType, replyTo, correlationId);
119128
} catch (IOException e) {
120129
return handleErrorResponse(e, replyTo, correlationId);
@@ -123,12 +132,14 @@ private Message<ResponseBody> getResponseBodyMessage(byte[] replyTo, byte[] corr
123132

124133
private Message<ResponseBody> handleErrorResponse(Exception e, byte[] replyTo, byte[] correlationId) {
125134
ResponseBody responseBody = ResponseBody.builder()
126-
.errorMessage(Collections.singletonMap(e.getClass().toString(), e.getMessage()))
127-
.build();
135+
.errorMessage(
136+
Collections.singletonMap(e.getClass().toString(), e.getMessage()))
137+
.build();
128138
return buildResponseMessage(responseBody, ResponseType.ERROR_RESPONSE, replyTo, correlationId);
129139
}
130140

131-
private Message<ResponseBody> buildResponseMessage(ResponseBody body, ResponseType responseType, byte[] replyTo, byte[] correlationId) {
141+
private Message<ResponseBody> buildResponseMessage(ResponseBody body, ResponseType responseType, byte[] replyTo,
142+
byte[] correlationId) {
132143
MessageHeaderAccessor accessor = new MessageHeaderAccessor();
133144
accessor.setHeader(KafkaHeaders.MESSAGE_KEY, responseType.name());
134145
accessor.setHeader(KafkaHeaders.CORRELATION_ID, correlationId);

0 commit comments

Comments
 (0)