ambari-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From oleew...@apache.org
Subject [1/2] ambari git commit: AMBARI-15887. Added Logfeeder unit tests (Miklos Gergely via oleewere)
Date Fri, 15 Apr 2016 19:17:46 GMT
Repository: ambari
Updated Branches:
  refs/heads/trunk a55c055a8 -> e5c13a3ad


http://git-wip-us.apache.org/repos/asf/ambari/blob/e5c13a3a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputKafkaTest.java
----------------------------------------------------------------------
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputKafkaTest.java
b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputKafkaTest.java
new file mode 100644
index 0000000..a7db3f8
--- /dev/null
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputKafkaTest.java
@@ -0,0 +1,128 @@
+/**
+ * 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.ambari.logfeeder.output;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.Future;
+
+import org.apache.ambari.logfeeder.input.Input;
+import org.apache.ambari.logfeeder.input.InputMarker;
+import org.apache.ambari.logfeeder.output.OutputKafka.KafkaCallBack;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.log4j.Logger;
+import org.easymock.EasyMock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class OutputKafkaTest {
+  private static final Logger LOG = Logger.getLogger(OutputKafkaTest.class);
+
+  private static final String TEST_TOPIC = "test topic";
+
+  private OutputKafka outputKafka;
+
+  @SuppressWarnings("unchecked")
+  private KafkaProducer<String, String> mockKafkaProducer = EasyMock.strictMock(KafkaProducer.class);
+
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Before
+  public void init() {
+    outputKafka = new OutputKafka() {
+      @Override
+      protected KafkaProducer<String, String> creteKafkaProducer(Properties props)
{
+        return mockKafkaProducer;
+      }
+    };
+  }
+
+  @Test
+  public void testOutputKafka_uploadData() throws Exception {
+    LOG.info("testOutputKafka_uploadData()");
+
+    Map<String, Object> config = new HashMap<String, Object>();
+    config.put("broker_list", "some broker list");
+    config.put("topic", TEST_TOPIC);
+
+    outputKafka.loadConfig(config);
+    outputKafka.init();
+
+    @SuppressWarnings("unchecked")
+    Future<RecordMetadata> mockFuture = EasyMock.mock(Future.class);
+
+    EasyMock.expect(mockKafkaProducer.send(new ProducerRecord<String, String>(TEST_TOPIC,
"value0")))
+        .andReturn(mockFuture);
+    EasyMock.expect(mockFuture.get()).andReturn(null);
+
+    for (int i = 1; i < 10; i++)
+      EasyMock.expect(mockKafkaProducer.send(EasyMock.eq(new ProducerRecord<String, String>(TEST_TOPIC,
"value" + i)),
+          EasyMock.anyObject(KafkaCallBack.class))).andReturn(null);
+
+    EasyMock.replay(mockKafkaProducer);
+
+    for (int i = 0; i < 10; i++) {
+      InputMarker inputMarker = new InputMarker();
+      inputMarker.input = EasyMock.mock(Input.class);
+      outputKafka.write("value" + i, inputMarker);
+    }
+
+    EasyMock.verify(mockKafkaProducer);
+  }
+
+  @Test
+  public void testOutputKafka_noBrokerList() throws Exception {
+    LOG.info("testOutputKafka_noBrokerList()");
+
+    expectedException.expect(Exception.class);
+    expectedException.expectMessage("For kafka output, bootstrap broker_list is needed");
+
+    Map<String, Object> config = new HashMap<String, Object>();
+    config.put("topic", TEST_TOPIC);
+
+    outputKafka.loadConfig(config);
+    outputKafka.init();
+  }
+
+  @Test
+  public void testOutputKafka_noTopic() throws Exception {
+    LOG.info("testOutputKafka_noBrokerList()");
+
+    expectedException.expect(Exception.class);
+    expectedException.expectMessage("For kafka output, topic is needed");
+
+    Map<String, Object> config = new HashMap<String, Object>();
+    config.put("broker_list", "some broker list");
+
+    outputKafka.loadConfig(config);
+    outputKafka.init();
+  }
+
+  @After
+  public void cleanUp() {
+    EasyMock.reset(mockKafkaProducer);
+  }
+}

http://git-wip-us.apache.org/repos/asf/ambari/blob/e5c13a3a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputSolrTest.java
----------------------------------------------------------------------
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputSolrTest.java
b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputSolrTest.java
new file mode 100644
index 0000000..afbccca
--- /dev/null
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/java/org/apache/ambari/logfeeder/output/OutputSolrTest.java
@@ -0,0 +1,165 @@
+/**
+ * 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.ambari.logfeeder.output;
+
+import java.net.MalformedURLException;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.ambari.logfeeder.input.Input;
+import org.apache.ambari.logfeeder.input.InputMarker;
+import org.apache.log4j.Logger;
+import org.apache.solr.client.solrj.SolrClient;
+import org.apache.solr.client.solrj.impl.CloudSolrClient;
+import org.apache.solr.client.solrj.response.UpdateResponse;
+import org.apache.solr.common.SolrInputDocument;
+import org.apache.solr.common.util.NamedList;
+import org.easymock.EasyMock;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class OutputSolrTest {
+  private static final Logger LOG = Logger.getLogger(OutputSolrTest.class);
+
+  private OutputSolr outputSolr;
+  private Map<Integer, SolrInputDocument> receivedDocs = new ConcurrentHashMap<>();
+
+  @Rule
+  public ExpectedException expectedException = ExpectedException.none();
+
+  @Before
+  public void init() throws Exception {
+    outputSolr = new OutputSolr() {
+      @Override
+      SolrClient getSolrClient(String solrUrl, String zkHosts, int count) throws Exception,
MalformedURLException {
+        return new CloudSolrClient(null) {
+          private static final long serialVersionUID = 1L;
+
+          @Override
+          public UpdateResponse add(Collection<SolrInputDocument> docs) {
+            for (SolrInputDocument doc : docs) {
+              receivedDocs.put((Integer) doc.getField("id").getValue(), doc);
+            }
+
+            UpdateResponse response = new UpdateResponse();
+            response.setResponse(new NamedList<Object>());
+            return response;
+          }
+        };
+      }
+    };
+  }
+
+  @Test
+  public void testOutputToSolr_uploadData() throws Exception {
+    LOG.info("testOutputToSolr_uploadData()");
+
+    Map<String, Object> config = new HashMap<String, Object>();
+    config.put("url", "some url");
+    config.put("workers", "3");
+
+    outputSolr.loadConfig(config);
+    outputSolr.init();
+
+    Map<Integer, SolrInputDocument> expectedDocs = new HashMap<>();
+
+    int count = 0;
+    for (int i = 0; i < 10; i++) {
+      Map<String, Object> jsonObj = new HashMap<>();
+      for (int j = 0; j < 3; j++)
+        jsonObj.put("name" + ++count, "value" + ++count);
+      jsonObj.put("id", ++count);
+
+      InputMarker inputMarker = new InputMarker();
+      inputMarker.input = EasyMock.mock(Input.class);
+      outputSolr.write(jsonObj, inputMarker);
+
+      SolrInputDocument doc = new SolrInputDocument();
+      for (Map.Entry<String, Object> e : jsonObj.entrySet())
+        doc.addField(e.getKey(), e.getValue());
+
+      expectedDocs.put(count, doc);
+    }
+
+    Thread.sleep(100);
+    while (outputSolr.getPendingCount() > 0)
+      Thread.sleep(100);
+
+    int waitToFinish = 0;
+    if (receivedDocs.size() < 10 && waitToFinish < 10) {
+      Thread.sleep(100);
+      waitToFinish++;
+    }
+
+    Set<Integer> ids = new HashSet<>();
+    ids.addAll(receivedDocs.keySet());
+    ids.addAll(expectedDocs.keySet());
+    for (Integer id : ids) {
+      SolrInputDocument receivedDoc = receivedDocs.get(id);
+      SolrInputDocument expectedDoc = expectedDocs.get(id);
+
+      assertNotNull("No document received for id: " + id, receivedDoc);
+      assertNotNull("No document expected for id: " + id, expectedDoc);
+
+      Set<String> fieldNames = new HashSet<>();
+      fieldNames.addAll(receivedDoc.getFieldNames());
+      fieldNames.addAll(expectedDoc.getFieldNames());
+
+      for (String fieldName : fieldNames) {
+        Object receivedValue = receivedDoc.getFieldValue(fieldName);
+        Object expectedValue = expectedDoc.getFieldValue(fieldName);
+
+        assertNotNull("No received document field found for id: " + id + ", fieldName: "
+ fieldName, receivedValue);
+        assertNotNull("No expected document field found for id: " + id + ", fieldName: "
+ fieldName, expectedValue);
+
+        assertEquals("Field value not matching for id: " + id + ", fieldName: " + fieldName,
receivedValue,
+            expectedValue);
+      }
+    }
+  }
+
+  @Test
+  public void testOutputToSolr_noUrlOrZKHost() throws Exception {
+    LOG.info("testOutputToSolr_noUrlOrZKHost()");
+
+    expectedException.expect(Exception.class);
+    expectedException.expectMessage("For solr output, either url or zk_hosts property need
to be set");
+
+    Map<String, Object> config = new HashMap<String, Object>();
+    config.put("workers", "3");
+
+    outputSolr.loadConfig(config);
+    outputSolr.init();
+  }
+
+  @After
+  public void cleanUp() {
+    receivedDocs.clear();
+  }
+}

http://git-wip-us.apache.org/repos/asf/ambari/blob/e5c13a3a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/resources/log4j.xml
----------------------------------------------------------------------
diff --git a/ambari-logsearch/ambari-logsearch-logfeeder/src/test/resources/log4j.xml b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/resources/log4j.xml
new file mode 100644
index 0000000..e641018
--- /dev/null
+++ b/ambari-logsearch/ambari-logsearch-logfeeder/src/test/resources/log4j.xml
@@ -0,0 +1,53 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!--
+  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.
+-->
+<!DOCTYPE log4j:configuration SYSTEM "log4j.dtd">
+
+<log4j:configuration xmlns:log4j="http://jakarta.apache.org/log4j/">
+  <appender name="console" class="org.apache.log4j.ConsoleAppender">
+    <param name="Target" value="System.out" />
+    <layout class="org.apache.log4j.PatternLayout">
+      <param name="ConversionPattern" value="%d [%t] %-5p %C{6} (%F:%L) - %m%n" />
+      <!-- <param name="ConversionPattern" value="%d [%t] %-5p %c %x - %m%n"/> -->
+    </layout>
+  </appender>
+
+  <!-- Logs to suppress BEGIN -->
+  <category name="org.apache.solr.common.cloud.ZkStateReader" additivity="false">
+    <priority value="error" />
+    <appender-ref ref="console" />
+  </category>
+
+  <category name="apache.solr.client.solrj.impl.CloudSolrClient" additivity="false">
+    <priority value="fatal" />
+    <appender-ref ref="console" />
+  </category>
+  <!-- Logs to suppress END -->
+
+  <category name="org.apache.ambari.logfeeder" additivity="false">
+    <priority value="info" />
+    <appender-ref ref="console" /> 
+    <!-- <appender-ref ref="daily_rolling_file" /> -->
+  </category>
+
+  <root>
+    <priority value="warn" />
+    <!-- <appender-ref ref="console" /> -->
+    <!-- <appender-ref ref="daily_rolling_file" /> -->
+  </root>
+ 
+</log4j:configuration>  


Mime
View raw message