You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by di...@apache.org on 2018/04/05 04:35:29 UTC
[airavata] branch develop updated: Synch commit support for kafka
job monitor queue
This is an automated email from the ASF dual-hosted git repository.
dimuthuupe pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/airavata.git
The following commit(s) were added to refs/heads/develop by this push:
new d042fd3 Synch commit support for kafka job monitor queue
d042fd3 is described below
commit d042fd3c5d1f003bb64efd2d6c7fb3fd4496ed7b
Author: dimuthu <di...@gmail.com>
AuthorDate: Thu Apr 5 00:35:20 2018 -0400
Synch commit support for kafka job monitor queue
---
.../airavata/api/server/util/ThriftClientPool.java | 176 ---------------------
.../helix/impl/workflow/PostWorkflowManager.java | 31 +++-
2 files changed, 23 insertions(+), 184 deletions(-)
diff --git a/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/util/ThriftClientPool.java b/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/util/ThriftClientPool.java
deleted file mode 100644
index 5843497..0000000
--- a/airavata-api/airavata-api-server/src/main/java/org/apache/airavata/api/server/util/ThriftClientPool.java
+++ /dev/null
@@ -1,176 +0,0 @@
-/**
- *
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.apache.airavata.api.server.util;
-
-import org.apache.commons.pool.BasePoolableObjectFactory;
-import org.apache.commons.pool.impl.GenericObjectPool;
-import org.apache.thrift.TServiceClient;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.TTransportException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-public class ThriftClientPool<T extends TServiceClient> implements
- AutoCloseable {
-
- private static final Logger LOGGER = LoggerFactory
- .getLogger(ThriftClientPool.class);
-
- private final GenericObjectPool internalPool;
-
- public ThriftClientPool(ClientFactory<T> clientFactory,
- GenericObjectPool.Config poolConfig, String host, int port) {
- this(clientFactory, new BinaryOverSocketProtocolFactory(host, port),
- poolConfig);
- }
-
- public ThriftClientPool(ClientFactory<T> clientFactory,
- ProtocolFactory protocolFactory, GenericObjectPool.Config poolConfig) {
- this.internalPool = new GenericObjectPool(new ThriftClientFactory(
- clientFactory, protocolFactory), poolConfig);
- }
-
- class ThriftClientFactory extends BasePoolableObjectFactory {
-
- private ClientFactory<T> clientFactory;
- private ProtocolFactory protocolFactory;
-
- public ThriftClientFactory(ClientFactory<T> clientFactory,
- ProtocolFactory protocolFactory) {
- this.clientFactory = clientFactory;
- this.protocolFactory = protocolFactory;
- }
-
- @Override
- public T makeObject() throws Exception {
- try {
- TProtocol protocol = protocolFactory.make();
- return clientFactory.make(protocol);
- } catch (Exception e) {
- LOGGER.warn(e.getMessage(), e);
- throw new ThriftClientException(
- "Can not make a new object for pool", e);
- }
- }
-
- public void destroyObject(T obj) throws Exception {
- if (obj.getOutputProtocol().getTransport().isOpen()) {
- obj.getOutputProtocol().getTransport().close();
- }
- if (obj.getInputProtocol().getTransport().isOpen()) {
- obj.getInputProtocol().getTransport().close();
- }
- }
- }
-
- public static interface ClientFactory<T> {
-
- T make(TProtocol tProtocol);
- }
-
- public static interface ProtocolFactory {
-
- TProtocol make();
- }
-
- public static class BinaryOverSocketProtocolFactory implements
- ProtocolFactory {
-
- private String host;
- private int port;
-
- public BinaryOverSocketProtocolFactory(String host, int port) {
- this.host = host;
- this.port = port;
- }
-
- public TProtocol make() {
- TTransport transport = new TSocket(host, port);
- try {
- transport.open();
- } catch (TTransportException e) {
- LOGGER.warn(e.getMessage(), e);
- throw new ThriftClientException("Can not make protocol", e);
- }
- return new TBinaryProtocol(transport);
- }
- }
-
- public static class ThriftClientException extends RuntimeException {
-
- // Fucking Eclipse
- private static final long serialVersionUID = -2275296727467192665L;
-
- public ThriftClientException(String message, Exception e) {
- super(message, e);
- }
-
- }
-
- public T getResource() {
- try {
- return (T) internalPool.borrowObject();
- } catch (Exception e) {
- throw new ThriftClientException(
- "Could not get a resource from the pool", e);
- }
- }
-
- public void returnResourceObject(T resource) {
- try {
- internalPool.returnObject(resource);
- } catch (Exception e) {
- throw new ThriftClientException(
- "Could not return the resource to the pool", e);
- }
- }
-
- public void returnBrokenResource(T resource) {
- returnBrokenResourceObject(resource);
- }
-
- public void returnResource(T resource) {
- returnResourceObject(resource);
- }
-
- protected void returnBrokenResourceObject(T resource) {
- try {
- internalPool.invalidateObject(resource);
- } catch (Exception e) {
- throw new ThriftClientException(
- "Could not return the resource to the pool", e);
- }
- }
-
- public void destroy() {
- close();
- }
-
- public void close() {
- try {
- internalPool.close();
- } catch (Exception e) {
- throw new ThriftClientException("Could not destroy the pool", e);
- }
- }
-}
\ No newline at end of file
diff --git a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
index 21290c7..0bf5eb8 100644
--- a/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
+++ b/modules/airavata-helix/helix-spectator/src/main/java/org/apache/airavata/helix/impl/workflow/PostWorkflowManager.java
@@ -43,10 +43,8 @@ import org.apache.airavata.model.task.DataStagingTaskModel;
import org.apache.airavata.model.task.TaskModel;
import org.apache.airavata.model.task.TaskTypes;
import org.apache.airavata.registry.api.RegistryService;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.consumer.ConsumerRecords;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;
@@ -73,6 +71,7 @@ public class PostWorkflowManager extends WorkflowManager {
props.put(ConsumerConfig.GROUP_ID_CONFIG, ServerSettings.getSetting("kafka.broker.consumer.group"));
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JobStatusResultDeserializer.class.getName());
+ props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
// Create the consumer using props.
final Consumer<String, JobStatusResult> consumer = new KafkaConsumer<>(props);
// Subscribe to the topic.
@@ -158,10 +157,11 @@ public class PostWorkflowManager extends WorkflowManager {
return stat != null;
}
- private void process(JobStatusResult jobStatusResult) {
+ private boolean process(JobStatusResult jobStatusResult) {
if (jobStatusResult == null) {
- return;
+ logger.error("Job result is null");
+ return false;
}
try {
@@ -173,7 +173,7 @@ public class PostWorkflowManager extends WorkflowManager {
if (!JobStateValidator.isValid(currentJobStatus, jobStatusResult.getState())) {
logger.warn("Job state of " + jobStatusResult.getJobId() + " is not valid. Previous state " +
currentJobStatus + ", new state " + jobStatusResult.getState());
- return;
+ return true;
}
String gateway = Optional.ofNullable(getGatewayByJobId(jobStatusResult.getJobId()))
@@ -295,11 +295,14 @@ public class PostWorkflowManager extends WorkflowManager {
}
}
+ return true;
} else {
logger.warn("Could not find a monitoring register for job id " + jobStatusResult.getJobId());
+ return false;
}
} catch (Exception e) {
logger.error("Failed to process job : " + jobStatusResult.getJobId() + ", with status : " + jobStatusResult.getState().name(), e);
+ return false;
}
}
@@ -307,7 +310,19 @@ public class PostWorkflowManager extends WorkflowManager {
final Consumer<String, JobStatusResult> consumer = createConsumer();
while (true) {
- final ConsumerRecords<String, JobStatusResult> consumerRecords = consumer.poll(1000);
+ final ConsumerRecords<String, JobStatusResult> consumerRecords = consumer.poll(Long.MAX_VALUE);
+
+ for (TopicPartition partition : consumerRecords.partitions()) {
+ List<ConsumerRecord<String, JobStatusResult>> partitionRecords = consumerRecords.records(partition);
+ for (ConsumerRecord<String, JobStatusResult> record : partitionRecords) {
+ boolean success = process(record.value());
+ logger.info("Status of processing " + record.value().getJobId() + " : " + success);
+ if (success) {
+ consumer.commitSync(Collections.singletonMap(partition, new OffsetAndMetadata(record.offset() + 1)));
+ }
+ }
+ }
+
consumerRecords.forEach(record -> {
process(record.value());
});
--
To stop receiving notification emails like this one, please contact
dimuthuupe@apache.org.