atlas-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From mad...@apache.org
Subject incubator-atlas git commit: ATLAS-1908: updated to use existing Kafka consumer properties when equivalent new Kafka consumer properties are not present
Date Fri, 14 Jul 2017 22:44:50 GMT
Repository: incubator-atlas
Updated Branches:
  refs/heads/master de9160844 -> 0d8f9f8d2


ATLAS-1908: updated to use existing Kafka consumer properties when equivalent new Kafka consumer
properties are not present

Signed-off-by: Madhan Neethiraj <madhan@apache.org>


Project: http://git-wip-us.apache.org/repos/asf/incubator-atlas/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-atlas/commit/0d8f9f8d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-atlas/tree/0d8f9f8d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-atlas/diff/0d8f9f8d

Branch: refs/heads/master
Commit: 0d8f9f8d2d3902471dc246bd51d3eb7512021f74
Parents: de91608
Author: nixonrodrigues <nixon@apache.org>
Authored: Mon Jul 3 18:18:47 2017 +0530
Committer: Madhan Neethiraj <madhan@apache.org>
Committed: Fri Jul 14 15:28:58 2017 -0700

----------------------------------------------------------------------
 distro/src/conf/atlas-application.properties        |  2 +-
 .../org/apache/atlas/kafka/AtlasKafkaConsumer.java  | 13 +++++++------
 .../org/apache/atlas/kafka/KafkaNotification.java   | 12 ++++++++++--
 .../atlas/notification/NotificationConsumer.java    |  3 +--
 .../org/apache/atlas/kafka/KafkaConsumerTest.java   | 16 ++++++++--------
 .../apache/atlas/kafka/KafkaNotificationTest.java   |  2 +-
 .../AbstractNotificationConsumerTest.java           |  8 ++++----
 .../notification/NotificationHookConsumer.java      |  2 +-
 .../NotificationHookConsumerKafkaTest.java          |  2 +-
 .../atlas/web/integration/BaseResourceIT.java       |  2 +-
 10 files changed, 35 insertions(+), 27 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/distro/src/conf/atlas-application.properties
----------------------------------------------------------------------
diff --git a/distro/src/conf/atlas-application.properties b/distro/src/conf/atlas-application.properties
index c3213df..29a4cc1 100755
--- a/distro/src/conf/atlas-application.properties
+++ b/distro/src/conf/atlas-application.properties
@@ -79,7 +79,7 @@ atlas.kafka.hook.group.id=atlas
 atlas.kafka.enable.auto.commit=false
 atlas.kafka.auto.offset.reset=earliest
 atlas.kafka.session.timeout.ms=30000
-
+atlas.kafka.poll.timeout.ms=1000
 
 atlas.notification.create.topics=true
 atlas.notification.replicas=1

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java b/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
index 9c15243..52d0916 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/AtlasKafkaConsumer.java
@@ -41,19 +41,20 @@ public class AtlasKafkaConsumer<T> extends AbstractNotificationConsumer<T>
{
     private static final Logger LOG = LoggerFactory.getLogger(AtlasKafkaConsumer.class);
 
     private final KafkaConsumer kafkaConsumer;
-    private final boolean       autoCommitEnabled;
+    private final boolean autoCommitEnabled;
+    private long pollTimeoutMilliSeconds = 1000L;
 
-    public AtlasKafkaConsumer(MessageDeserializer<T> deserializer, KafkaConsumer kafkaConsumer,
boolean autoCommitEnabled) {
+    public AtlasKafkaConsumer(MessageDeserializer<T> deserializer, KafkaConsumer kafkaConsumer,
boolean autoCommitEnabled, long pollTimeoutMilliSeconds) {
         super(deserializer);
-
-        this.kafkaConsumer     = kafkaConsumer;
+        this.kafkaConsumer = kafkaConsumer;
         this.autoCommitEnabled = autoCommitEnabled;
+        this.pollTimeoutMilliSeconds = pollTimeoutMilliSeconds;
     }
 
-    public List<AtlasKafkaMessage<T>> receive(long timeoutMilliSeconds) {
+    public List<AtlasKafkaMessage<T>> receive() {
         List<AtlasKafkaMessage<T>> messages = new ArrayList();
 
-        ConsumerRecords<?, ?> records = kafkaConsumer.poll(timeoutMilliSeconds);
+        ConsumerRecords<?, ?> records = kafkaConsumer.poll(pollTimeoutMilliSeconds);
 
         if (records != null) {
             for (ConsumerRecord<?, ?> record : records) {

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
index 366c8a7..38889ef 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -83,6 +83,7 @@ public class KafkaNotification extends AbstractNotification implements Service
{
     private Properties properties;
     private KafkaConsumer consumer = null;
     private KafkaProducer producer = null;
+    private Long pollTimeOutMs = 1000L;
 
     private static final Map<NotificationType, String> TOPIC_MAP = new HashMap<NotificationType,
String>() {
         {
@@ -124,6 +125,13 @@ public class KafkaNotification extends AbstractNotification implements
Service {
         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                 "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
+
+        pollTimeOutMs = subsetConfiguration.getLong("poll.timeout.ms", 1000);
+        boolean oldApiCommitEnbleFlag = subsetConfiguration.getBoolean("auto.commit.enable",false);
+        //set old autocommit value if new autoCommit property is not set.
+        properties.put("enable.auto.commit", subsetConfiguration.getBoolean("enable.auto.commit",
oldApiCommitEnbleFlag));
+        properties.put("session.timeout.ms", subsetConfiguration.getString("session.timeout.ms",
"30000"));
+
     }
 
     @VisibleForTesting
@@ -167,7 +175,7 @@ public class KafkaNotification extends AbstractNotification implements
Service {
     public <T> List<NotificationConsumer<T>> createConsumers(NotificationType
notificationType,
                                                              int numConsumers) {
         return createConsumers(notificationType, numConsumers,
-                Boolean.valueOf(properties.getProperty("enable.auto.commit", "true")));
+                Boolean.valueOf(properties.getProperty("enable.auto.commit", properties.getProperty("auto.commit.enable","false"))));
     }
 
     @VisibleForTesting
@@ -177,7 +185,7 @@ public class KafkaNotification extends AbstractNotification implements
Service {
         Properties consumerProperties = getConsumerProperties(notificationType);
 
         List<NotificationConsumer<T>> consumers = new ArrayList<>();
-        AtlasKafkaConsumer kafkaConsumer = new AtlasKafkaConsumer(notificationType.getDeserializer(),
getKafkaConsumer(consumerProperties,notificationType, autoCommitEnabled), autoCommitEnabled);
+        AtlasKafkaConsumer kafkaConsumer = new AtlasKafkaConsumer(notificationType.getDeserializer(),
getKafkaConsumer(consumerProperties,notificationType, autoCommitEnabled), autoCommitEnabled,
pollTimeOutMs );
         consumers.add(kafkaConsumer);
         return consumers;
     }

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/main/java/org/apache/atlas/notification/NotificationConsumer.java
----------------------------------------------------------------------
diff --git a/notification/src/main/java/org/apache/atlas/notification/NotificationConsumer.java
b/notification/src/main/java/org/apache/atlas/notification/NotificationConsumer.java
index 22e40f9..6d1c08a 100644
--- a/notification/src/main/java/org/apache/atlas/notification/NotificationConsumer.java
+++ b/notification/src/main/java/org/apache/atlas/notification/NotificationConsumer.java
@@ -41,8 +41,7 @@ public interface NotificationConsumer<T> {
 
     /**
      * Fetch data for the topics from Kafka
-     * @param timeoutMilliSeconds poll timeout
      * @return List containing kafka message and partionId and offset.
      */
-    List<AtlasKafkaMessage<T>> receive(long timeoutMilliSeconds);
+    List<AtlasKafkaMessage<T>> receive();
 }

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java b/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
index 70059cb..9b712f4 100644
--- a/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
+++ b/notification/src/test/java/org/apache/atlas/kafka/KafkaConsumerTest.java
@@ -95,12 +95,12 @@ public class KafkaConsumerTest {
         ConsumerRecords records = new ConsumerRecords(mp);
 
 
-        when(kafkaConsumer.poll(1000)).thenReturn(records);
+        when(kafkaConsumer.poll(100)).thenReturn(records);
         when(messageAndMetadata.message()).thenReturn(json);
 
 
-        AtlasKafkaConsumer consumer = new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer,false);
-        List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>> messageList
= consumer.receive(1000);
+        AtlasKafkaConsumer consumer = new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer, false, 100L);
+        List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>> messageList
= consumer.receive();
         assertTrue(messageList.size() > 0);
 
         HookNotification.HookNotificationMessage consumedMessage  = messageList.get(0).getMessage();
@@ -131,12 +131,12 @@ public class KafkaConsumerTest {
         mp.put(tp,klist);
         ConsumerRecords records = new ConsumerRecords(mp);
 
-        when(kafkaConsumer.poll(1000)).thenReturn(records);
+        when(kafkaConsumer.poll(100L)).thenReturn(records);
         when(messageAndMetadata.message()).thenReturn(json);
 
-        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer ,false);
+        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer ,false, 100L);
         try {
-            List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>>
messageList = consumer.receive(1000);
+            List<AtlasKafkaMessage<HookNotification.HookNotificationMessage>>
messageList = consumer.receive();
             assertTrue(messageList.size() > 0);
 
             HookNotification.HookNotificationMessage consumedMessage  = messageList.get(0).getMessage();
@@ -154,7 +154,7 @@ public class KafkaConsumerTest {
 
         TopicPartition tp = new TopicPartition("ATLAS_HOOK",0);
 
-        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer, false);
+        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer, false, 100L);
 
         consumer.commit(tp, 1);
 
@@ -166,7 +166,7 @@ public class KafkaConsumerTest {
 
         TopicPartition tp = new TopicPartition("ATLAS_HOOK",0);
 
-        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer, true);
+        AtlasKafkaConsumer consumer =new AtlasKafkaConsumer(NotificationInterface.NotificationType.HOOK.getDeserializer(),
kafkaConsumer, true , 100L);
 
         consumer.commit(tp, 1);
 

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
b/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
index c791d43..a1e13b9 100644
--- a/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
+++ b/notification/src/test/java/org/apache/atlas/kafka/KafkaNotificationTest.java
@@ -70,7 +70,7 @@ public class KafkaNotificationTest {
         List<AtlasKafkaMessage<Object>> messages = null ;
         long startTime = System.currentTimeMillis(); //fetch starting time
         while ((System.currentTimeMillis() - startTime) < 10000) {
-             messages = consumer.receive(1000L);
+             messages = consumer.receive();
             if (messages.size() > 0) {
                 break;
             }

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
----------------------------------------------------------------------
diff --git a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
index 8324b57..68fe3d7 100644
--- a/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
+++ b/notification/src/test/java/org/apache/atlas/notification/AbstractNotificationConsumerTest.java
@@ -67,7 +67,7 @@ public class AbstractNotificationConsumerTest {
         NotificationConsumer<TestMessage> consumer =
                 new TestNotificationConsumer<>(versionedMessageType, jsonList, logger);
 
-        List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive(1000L);
+        List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
 
         assertFalse(messageList.isEmpty());
 
@@ -106,7 +106,7 @@ public class AbstractNotificationConsumerTest {
         NotificationConsumer<TestMessage> consumer =
             new TestNotificationConsumer<>(versionedMessageType, jsonList, logger);
 
-        List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive(1000L);
+        List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
 
         assertEquals(new TestMessage("sValue1", 99), messageList.get(0).getMessage());
 
@@ -138,7 +138,7 @@ public class AbstractNotificationConsumerTest {
         NotificationConsumer<TestMessage> consumer =
             new TestNotificationConsumer<>(versionedMessageType, jsonList, logger);
         try {
-            List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive(1000L);
+            List<AtlasKafkaMessage<TestMessage>> messageList = consumer.receive();
 
             messageList.get(1).getMessage();
 
@@ -203,7 +203,7 @@ public class AbstractNotificationConsumerTest {
         }
 
         @Override
-        public List<AtlasKafkaMessage<T>> receive(long timeoutMilliSeconds) {
+        public List<AtlasKafkaMessage<T>> receive() {
             List<AtlasKafkaMessage<T>> tempMessageList = new ArrayList();
             for(Object json :  messageList) {
                 tempMessageList.add(new AtlasKafkaMessage(deserializer.deserialize((String)json),
-1, -1));

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
----------------------------------------------------------------------
diff --git a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
index 0dea0e2..51276d3 100644
--- a/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
+++ b/webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java
@@ -224,7 +224,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
 
             while (shouldRun.get()) {
                 try {
-                    List<AtlasKafkaMessage<HookNotificationMessage>> messages
= consumer.receive(1000L);
+                    List<AtlasKafkaMessage<HookNotificationMessage>> messages
= consumer.receive();
                     for (AtlasKafkaMessage<HookNotificationMessage> msg : messages)
{
                         handleMessage(msg);
                     }

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
index 650ca0a..eb37fa8 100644
--- a/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
+++ b/webapp/src/test/java/org/apache/atlas/notification/NotificationHookConsumerKafkaTest.java
@@ -155,7 +155,7 @@ public class NotificationHookConsumerKafkaTest {
         try {
             long startTime = System.currentTimeMillis(); //fetch starting time
             while ((System.currentTimeMillis() - startTime) < 10000) {
-                List<AtlasKafkaMessage<HookNotificationMessage>> messages = consumer.receive(1000L);
+                List<AtlasKafkaMessage<HookNotificationMessage>> messages = consumer.receive();
 
                 for (AtlasKafkaMessage<HookNotificationMessage> msg : messages) {
                     hookConsumer.handleMessage(msg);

http://git-wip-us.apache.org/repos/asf/incubator-atlas/blob/0d8f9f8d/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
----------------------------------------------------------------------
diff --git a/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java b/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
index c036cfa..496185f 100755
--- a/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/integration/BaseResourceIT.java
@@ -637,7 +637,7 @@ public abstract class BaseResourceIT {
                 try {
 
                     while (System.currentTimeMillis() < maxCurrentTime) {
-                        List<AtlasKafkaMessage<EntityNotification>> messageList
= consumer.receive(1000);
+                        List<AtlasKafkaMessage<EntityNotification>> messageList
= consumer.receive();
                             if(messageList.size() > 0) {
                                 EntityNotification notification = messageList.get(0).getMessage();
                                 if (predicate.evaluate(notification)) {


Mime
View raw message