You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by mp...@apache.org on 2016/07/08 22:50:58 UTC
[6/9] flume git commit: FLUME-2941. Integrate checkstyle for test
classes
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
index 7851536..8921a19 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannel.java
@@ -19,10 +19,6 @@
package org.apache.flume.channel;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.concurrent.LinkedBlockingDeque;
-
import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
@@ -34,8 +30,12 @@ import org.apache.flume.event.EventBuilder;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
-import static org.fest.reflect.core.Reflection.*;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingDeque;
+
+import static org.fest.reflect.core.Reflection.field;
public class TestMemoryChannel {
@@ -81,7 +81,7 @@ public class TestMemoryChannel {
Transaction transaction = channel.getTransaction();
transaction.begin();
- for(int i=0; i < 5; i++) {
+ for (int i = 0; i < 5; i++) {
channel.put(EventBuilder.withBody(String.format("test event %d", i).getBytes()));
}
transaction.commit();
@@ -124,7 +124,7 @@ public class TestMemoryChannel {
parms.put("transactionCapacity", "2");
context.putAll(parms);
Configurables.configure(channel, context);
- for(int i=0; i < 6; i++) {
+ for (int i = 0; i < 6; i++) {
transaction = channel.getTransaction();
transaction.begin();
Assert.assertNotNull(channel.take());
@@ -133,7 +133,7 @@ public class TestMemoryChannel {
}
}
- @Test(expected=ChannelException.class)
+ @Test(expected = ChannelException.class)
public void testTransactionPutCapacityOverload() {
Context context = new Context();
Map<String, String> parms = new HashMap<String, String>();
@@ -151,7 +151,7 @@ public class TestMemoryChannel {
Assert.fail();
}
- @Test(expected=ChannelException.class)
+ @Test(expected = ChannelException.class)
public void testCapacityOverload() {
Context context = new Context();
Map<String, String> parms = new HashMap<String, String>();
@@ -236,7 +236,7 @@ public class TestMemoryChannel {
tx.close();
}
- @Test(expected=ChannelException.class)
+ @Test(expected = ChannelException.class)
public void testByteCapacityOverload() {
Context context = new Context();
Map<String, String> parms = new HashMap<String, String>();
@@ -284,8 +284,7 @@ public class TestMemoryChannel {
try {
channel.put(EventBuilder.withBody(eventBody));
throw new RuntimeException("Put was able to overflow byte capacity.");
- } catch (ChannelException ce)
- {
+ } catch (ChannelException ce) {
//Do nothing
}
@@ -306,8 +305,7 @@ public class TestMemoryChannel {
try {
channel.put(EventBuilder.withBody(eventBody));
throw new RuntimeException("Put was able to overflow byte capacity.");
- } catch (ChannelException ce)
- {
+ } catch (ChannelException ce) {
//Do nothing
}
tx.commit();
@@ -370,7 +368,7 @@ public class TestMemoryChannel {
channel.put(EventBuilder.withBody(eventBody));
tx.commit();
Assert.fail();
- } catch ( ChannelException e ) {
+ } catch (ChannelException e) {
//success
tx.rollback();
} finally {
@@ -397,12 +395,12 @@ public class TestMemoryChannel {
tx = channel.getTransaction();
tx.begin();
try {
- for(int i = 0; i < 2; i++) {
+ for (int i = 0; i < 2; i++) {
channel.put(EventBuilder.withBody(eventBody));
}
tx.commit();
Assert.fail();
- } catch ( ChannelException e ) {
+ } catch (ChannelException e) {
//success
tx.rollback();
} finally {
@@ -418,12 +416,12 @@ public class TestMemoryChannel {
tx.begin();
try {
- for(int i = 0; i < 15; i++) {
+ for (int i = 0; i < 15; i++) {
channel.put(EventBuilder.withBody(eventBody));
}
tx.commit();
Assert.fail();
- } catch ( ChannelException e ) {
+ } catch (ChannelException e) {
//success
tx.rollback();
} finally {
@@ -438,12 +436,12 @@ public class TestMemoryChannel {
tx.begin();
try {
- for(int i = 0; i < 25; i++) {
+ for (int i = 0; i < 25; i++) {
channel.put(EventBuilder.withBody(eventBody));
}
tx.commit();
Assert.fail();
- } catch ( ChannelException e ) {
+ } catch (ChannelException e) {
//success
tx.rollback();
} finally {
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
index d4ba705..68aa117 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelConcurrency.java
@@ -17,15 +17,6 @@
*/
package org.apache.flume.channel;
-import java.util.Map.Entry;
-import java.util.Random;
-import java.util.concurrent.BrokenBarrierException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.CyclicBarrier;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
-
import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
@@ -37,6 +28,15 @@ import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import java.util.Map.Entry;
+import java.util.Random;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
public class TestMemoryChannelConcurrency {
private CyclicBarrier barrier;
@@ -120,10 +120,10 @@ public class TestMemoryChannelConcurrency {
}
/**
- * Works with a startgate/endgate latches to make sure all threads run at the same time. Threads randomly
- * choose to commit or rollback random numbers of actions, tagging them with the thread no.
- * The correctness check is made by recording committed entries into a map, and verifying the count
- * after the endgate
+ * Works with a startgate/endgate latches to make sure all threads run at the same time.
+ * Threads randomly choose to commit or rollback random numbers of actions, tagging them with the
+ * thread no. The correctness check is made by recording committed entries into a map, and
+ * verifying the count after the endgate.
* Since nothing is taking the puts out, allow for a big capacity
*
* @throws InterruptedException
@@ -135,7 +135,8 @@ public class TestMemoryChannelConcurrency {
context.put("keep-alive", "1");
context.put("capacity", "5000"); // theoretical maximum of 100 threads * 10 * 5
// because we're just grabbing the whole lot in one commit
- // normally a transactionCapacity significantly lower than the channel capacity would be recommended
+ // normally a transactionCapacity significantly lower than the channel capacity would be
+ // recommended
context.put("transactionCapacity", "5000");
Configurables.configure(channel, context);
final ConcurrentHashMap<String, AtomicInteger> committedPuts =
@@ -158,17 +159,17 @@ public class TestMemoryChannelConcurrency {
} catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
- for(int j = 0; j < 10; j++) {
+ for (int j = 0; j < 10; j++) {
int events = rng.nextInt(5) + 1;
Transaction tx = channel.getTransaction();
tx.begin();
- for(int k = 0; k < events; k++) {
+ for (int k = 0; k < events; k++) {
channel.put(EventBuilder.withBody(strtid.getBytes()));
}
- if(rng.nextBoolean()) {
+ if (rng.nextBoolean()) {
tx.commit();
AtomicInteger tcount = committedPuts.get(strtid);
- if(tcount == null) {
+ if (tcount == null) {
committedPuts.put(strtid, new AtomicInteger(events));
} else {
tcount.addAndGet(events);
@@ -186,7 +187,7 @@ public class TestMemoryChannelConcurrency {
startGate.countDown();
endGate.await();
- if(committedPuts.isEmpty()) {
+ if (committedPuts.isEmpty()) {
Assert.fail();
}
@@ -194,17 +195,17 @@ public class TestMemoryChannelConcurrency {
Transaction tx = channel.getTransaction();
tx.begin();
Event e;
- while((e = channel.take()) != null) {
+ while ((e = channel.take()) != null) {
String index = new String(e.getBody());
AtomicInteger remain = committedPuts.get(index);
int post = remain.decrementAndGet();
- if(post == 0) {
+ if (post == 0) {
committedPuts.remove(index);
}
}
tx.commit();
tx.close();
- if(!committedPuts.isEmpty()) {
+ if (!committedPuts.isEmpty()) {
Assert.fail();
}
}
@@ -216,10 +217,12 @@ public class TestMemoryChannelConcurrency {
context.put("keep-alive", "1");
context.put("capacity", "100"); // theoretical maximum of 100 threads * 10 * 5
// because we're just grabbing the whole lot in one commit
- // normally a transactionCapacity significantly lower than the channel capacity would be recommended
+ // normally a transactionCapacity significantly lower than the channel capacity would be
+ // recommended
context.put("transactionCapacity", "100");
Configurables.configure(channel, context);
- final ConcurrentHashMap<String, AtomicInteger> committedPuts = new ConcurrentHashMap<String, AtomicInteger>();
+ final ConcurrentHashMap<String, AtomicInteger> committedPuts =
+ new ConcurrentHashMap<String, AtomicInteger>();
final ConcurrentHashMap<String, AtomicInteger> committedTakes =
new ConcurrentHashMap<String, AtomicInteger>();
@@ -228,7 +231,7 @@ public class TestMemoryChannelConcurrency {
final CountDownLatch endGate = new CountDownLatch(threadCount);
// start a sink and source for each
- for (int i = 0; i < threadCount/2; i++) {
+ for (int i = 0; i < threadCount / 2; i++) {
Thread t = new Thread() {
@Override
public void run() {
@@ -241,23 +244,23 @@ public class TestMemoryChannelConcurrency {
} catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
- for(int j = 0; j < 10; j++) {
+ for (int j = 0; j < 10; j++) {
int events = rng.nextInt(5) + 1;
Transaction tx = channel.getTransaction();
tx.begin();
- for(int k = 0; k < events; k++) {
+ for (int k = 0; k < events; k++) {
channel.put(EventBuilder.withBody(strtid.getBytes()));
}
- if(rng.nextBoolean()) {
+ if (rng.nextBoolean()) {
try {
tx.commit();
AtomicInteger tcount = committedPuts.get(strtid);
- if(tcount == null) {
+ if (tcount == null) {
committedPuts.put(strtid, new AtomicInteger(events));
} else {
tcount.addAndGet(events);
}
- } catch(ChannelException e) {
+ } catch (ChannelException e) {
System.out.print("puts commit failed");
tx.rollback();
}
@@ -282,25 +285,25 @@ public class TestMemoryChannelConcurrency {
} catch (InterruptedException e1) {
Thread.currentThread().interrupt();
}
- for(int j = 0; j < 10; j++) {
+ for (int j = 0; j < 10; j++) {
int events = rng.nextInt(5) + 1;
Transaction tx = channel.getTransaction();
tx.begin();
Event[] taken = new Event[events];
int k;
- for(k = 0; k < events; k++) {
+ for (k = 0; k < events; k++) {
taken[k] = channel.take();
- if(taken[k] == null) break;
+ if (taken[k] == null) break;
}
- if(rng.nextBoolean()) {
+ if (rng.nextBoolean()) {
try {
tx.commit();
- for(Event e : taken) {
- if(e == null) break;
+ for (Event e : taken) {
+ if (e == null) break;
String index = new String(e.getBody());
- synchronized(takeMapLock) {
+ synchronized (takeMapLock) {
AtomicInteger remain = committedTakes.get(index);
- if(remain == null) {
+ if (remain == null) {
committedTakes.put(index, new AtomicInteger(1));
} else {
remain.incrementAndGet();
@@ -323,7 +326,7 @@ public class TestMemoryChannelConcurrency {
t.start();
}
startGate.countDown();
- if(!endGate.await(20, TimeUnit.SECONDS)) {
+ if (!endGate.await(20, TimeUnit.SECONDS)) {
Assert.fail("Not all threads ended succesfully");
}
@@ -333,11 +336,11 @@ public class TestMemoryChannelConcurrency {
Event e;
// first pull out what's left in the channel and remove it from the
// committed map
- while((e = channel.take()) != null) {
+ while ((e = channel.take()) != null) {
String index = new String(e.getBody());
AtomicInteger remain = committedPuts.get(index);
int post = remain.decrementAndGet();
- if(post == 0) {
+ if (post == 0) {
committedPuts.remove(index);
}
}
@@ -345,14 +348,19 @@ public class TestMemoryChannelConcurrency {
tx.close();
// now just check the committed puts match the committed takes
- for(Entry<String, AtomicInteger> takes : committedTakes.entrySet()) {
+ for (Entry<String, AtomicInteger> takes : committedTakes.entrySet()) {
AtomicInteger count = committedPuts.get(takes.getKey());
- if(count == null)
+ if (count == null) {
Assert.fail("Putted data doesn't exist");
- if(count.get() != takes.getValue().get())
- Assert.fail(String.format("Mismatched put and take counts expected %d had %d", count.get(), takes.getValue().get()));
+ }
+ if (count.get() != takes.getValue().get()) {
+ Assert.fail(String.format("Mismatched put and take counts expected %d had %d",
+ count.get(), takes.getValue().get()));
+ }
committedPuts.remove(takes.getKey());
}
- if(!committedPuts.isEmpty()) Assert.fail("Puts still has entries remaining");
+ if (!committedPuts.isEmpty()) {
+ Assert.fail("Puts still has entries remaining");
+ }
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
index b8e00d8..55b81ee 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/channel/TestMemoryChannelTransaction.java
@@ -43,7 +43,8 @@ public class TestMemoryChannelTransaction {
@Test
public void testCommit() throws InterruptedException, EventDeliveryException {
- Event event, event2;
+ Event event;
+ Event event2;
Context context = new Context();
int putCounter = 0;
@@ -85,7 +86,8 @@ public class TestMemoryChannelTransaction {
public void testRollBack() throws InterruptedException,
EventDeliveryException {
- Event event, event2;
+ Event event;
+ Event event2;
Context context = new Context();
int putCounter = 0;
@@ -158,7 +160,8 @@ public class TestMemoryChannelTransaction {
public void testReEntTxn() throws InterruptedException,
EventDeliveryException {
- Event event, event2;
+ Event event;
+ Event event2;
Context context = new Context();
int putCounter = 0;
@@ -199,7 +202,8 @@ public class TestMemoryChannelTransaction {
@Test
public void testReEntTxnRollBack() throws InterruptedException,
EventDeliveryException {
- Event event, event2;
+ Event event;
+ Event event2;
Context context = new Context();
int putCounter = 0;
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestReliableSpoolingFileEventReader.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestReliableSpoolingFileEventReader.java b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestReliableSpoolingFileEventReader.java
index 4e90054..fdc3ce9 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestReliableSpoolingFileEventReader.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/client/avro/TestReliableSpoolingFileEventReader.java
@@ -37,14 +37,25 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileFilter;
import java.io.IOException;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
-import java.util.concurrent.*;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.Semaphore;
public class TestReliableSpoolingFileEventReader {
- private static final Logger logger = LoggerFactory.getLogger
- (TestReliableSpoolingFileEventReader.class);
+ private static final Logger logger =
+ LoggerFactory.getLogger(TestReliableSpoolingFileEventReader.class);
private static final File WORK_DIR = new File("target/test/work/" +
TestReliableSpoolingFileEventReader.class.getSimpleName());
@@ -57,7 +68,7 @@ public class TestReliableSpoolingFileEventReader {
// write out a few files
for (int i = 0; i < 4; i++) {
- File fileName = new File(WORK_DIR, "file"+i);
+ File fileName = new File(WORK_DIR, "file" + i);
StringBuilder sb = new StringBuilder();
// write as many lines as the index of the file
@@ -102,11 +113,12 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testIgnorePattern() throws IOException {
- ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .ignorePattern("^file2$")
- .deletePolicy(DeletePolicy.IMMEDIATE.toString())
- .build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder()
+ .spoolDirectory(WORK_DIR)
+ .ignorePattern("^file2$")
+ .deletePolicy(DeletePolicy.IMMEDIATE.toString())
+ .build();
List<File> before = listFiles(WORK_DIR);
Assert.assertEquals("Expected 5, not: " + before, 5, before.size());
@@ -128,8 +140,9 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testRepeatedCallsWithCommitAlways() throws IOException {
- ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR).build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .build();
final int expectedLines = 0 + 1 + 2 + 3 + 1;
int seenLines = 0;
@@ -148,8 +161,10 @@ public class TestReliableSpoolingFileEventReader {
SpoolDirectorySourceConfigurationConstants.DEFAULT_TRACKER_DIR;
File trackerDir = new File(WORK_DIR, trackerDirPath);
- ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR).trackerDirPath(trackerDirPath).build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .trackerDirPath(trackerDirPath)
+ .build();
final int expectedLines = 0 + 1 + 2 + 3 + 1;
int seenLines = 0;
@@ -173,10 +188,10 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testFileDeletion() throws IOException {
- ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .deletePolicy(DeletePolicy.IMMEDIATE.name())
- .build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .deletePolicy(DeletePolicy.IMMEDIATE.name())
+ .build();
List<File> before = listFiles(WORK_DIR);
Assert.assertEquals("Expected 5, not: " + before, 5, before.size());
@@ -197,29 +212,25 @@ public class TestReliableSpoolingFileEventReader {
@Test(expected = NullPointerException.class)
public void testNullConsumeOrder() throws IOException {
- new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(null)
- .build();
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(null)
+ .build();
}
@Test
public void testConsumeFileRandomly() throws IOException {
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.RANDOM)
- .build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.RANDOM)
+ .build();
File fileName = new File(WORK_DIR, "new-file");
- FileUtils.write(fileName,
- "New file created in the end. Shoud be read randomly.\n");
+ FileUtils.write(fileName, "New file created in the end. Shoud be read randomly.\n");
Set<String> actual = Sets.newHashSet();
readEventsForFilesInDir(WORK_DIR, reader, actual);
Set<String> expected = Sets.newHashSet();
createExpectedFromFilesInSetup(expected);
expected.add("");
- expected.add(
- "New file created in the end. Shoud be read randomly.");
+ expected.add("New file created in the end. Shoud be read randomly.");
Assert.assertEquals(expected, actual);
}
@@ -229,54 +240,46 @@ public class TestReliableSpoolingFileEventReader {
if (SystemUtils.IS_OS_WINDOWS) {
return;
}
- final ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.RANDOM)
- .build();
+ final ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.RANDOM)
+ .build();
File fileName = new File(WORK_DIR, "new-file");
- FileUtils.write(fileName,
- "New file created in the end. Shoud be read randomly.\n");
+ FileUtils.write(fileName, "New file created in the end. Shoud be read randomly.\n");
Set<String> expected = Sets.newHashSet();
int totalFiles = WORK_DIR.listFiles().length;
final Set<String> actual = Sets.newHashSet();
ExecutorService executor = Executors.newSingleThreadExecutor();
final Semaphore semaphore1 = new Semaphore(0);
final Semaphore semaphore2 = new Semaphore(0);
- Future<Void> wait = executor.submit(
- new Callable<Void>() {
- @Override
- public Void call() throws Exception {
- readEventsForFilesInDir(WORK_DIR, reader, actual, semaphore1, semaphore2);
- return null;
- }
+ Future<Void> wait = executor.submit(new Callable<Void>() {
+ @Override
+ public Void call() throws Exception {
+ readEventsForFilesInDir(WORK_DIR, reader, actual, semaphore1, semaphore2);
+ return null;
}
- );
+ });
semaphore1.acquire();
File finalFile = new File(WORK_DIR, "t-file");
FileUtils.write(finalFile, "Last file");
semaphore2.release();
wait.get();
- int listFilesCount = ((ReliableSpoolingFileEventReader)reader)
- .getListFilesCount();
+ int listFilesCount = ((ReliableSpoolingFileEventReader)reader).getListFilesCount();
finalFile.delete();
createExpectedFromFilesInSetup(expected);
expected.add("");
- expected.add(
- "New file created in the end. Shoud be read randomly.");
+ expected.add("New file created in the end. Shoud be read randomly.");
expected.add("Last file");
Assert.assertTrue(listFilesCount < (totalFiles + 2));
Assert.assertEquals(expected, actual);
}
-
@Test
public void testConsumeFileOldest() throws IOException, InterruptedException {
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.OLDEST)
- .build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.OLDEST)
+ .build();
File file1 = new File(WORK_DIR, "new-file1");
File file2 = new File(WORK_DIR, "new-file2");
File file3 = new File(WORK_DIR, "new-file3");
@@ -299,13 +302,11 @@ public class TestReliableSpoolingFileEventReader {
}
@Test
- public void testConsumeFileYoungest()
- throws IOException, InterruptedException {
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.YOUNGEST)
- .build();
+ public void testConsumeFileYoungest() throws IOException, InterruptedException {
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.YOUNGEST)
+ .build();
File file1 = new File(WORK_DIR, "new-file1");
File file2 = new File(WORK_DIR, "new-file2");
File file3 = new File(WORK_DIR, "new-file3");
@@ -332,12 +333,11 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testConsumeFileOldestWithLexicographicalComparision()
- throws IOException, InterruptedException {
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.OLDEST)
- .build();
+ throws IOException, InterruptedException {
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.OLDEST)
+ .build();
File file1 = new File(WORK_DIR, "new-file1");
File file2 = new File(WORK_DIR, "new-file2");
File file3 = new File(WORK_DIR, "new-file3");
@@ -362,12 +362,11 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testConsumeFileYoungestWithLexicographicalComparision()
- throws IOException, InterruptedException {
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR)
- .consumeOrder(ConsumeOrder.YOUNGEST)
- .build();
+ throws IOException, InterruptedException {
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .consumeOrder(ConsumeOrder.YOUNGEST)
+ .build();
File file1 = new File(WORK_DIR, "new-file1");
File file2 = new File(WORK_DIR, "new-file2");
File file3 = new File(WORK_DIR, "new-file3");
@@ -393,6 +392,7 @@ public class TestReliableSpoolingFileEventReader {
@Test public void testLargeNumberOfFilesOLDEST() throws IOException {
templateTestForLargeNumberOfFiles(ConsumeOrder.OLDEST, null, 1000);
}
+
@Test public void testLargeNumberOfFilesYOUNGEST() throws IOException {
templateTestForLargeNumberOfFiles(ConsumeOrder.YOUNGEST, new Comparator<Long>() {
@@ -402,6 +402,7 @@ public class TestReliableSpoolingFileEventReader {
}
}, 1000);
}
+
@Test public void testLargeNumberOfFilesRANDOM() throws IOException {
templateTestForLargeNumberOfFiles(ConsumeOrder.RANDOM, null, 1000);
}
@@ -409,19 +410,21 @@ public class TestReliableSpoolingFileEventReader {
@Test
public void testZeroByteTrackerFile() throws IOException {
String trackerDirPath =
- SpoolDirectorySourceConfigurationConstants.DEFAULT_TRACKER_DIR;
+ SpoolDirectorySourceConfigurationConstants.DEFAULT_TRACKER_DIR;
File trackerDir = new File(WORK_DIR, trackerDirPath);
- if(!trackerDir.exists()) {
+ if (!trackerDir.exists()) {
trackerDir.mkdir();
}
File trackerFile = new File(trackerDir, ReliableSpoolingFileEventReader.metaFileName);
- if(trackerFile.exists()) {
+ if (trackerFile.exists()) {
trackerFile.delete();
}
trackerFile.createNewFile();
- ReliableEventReader reader = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(WORK_DIR).trackerDirPath(trackerDirPath).build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(WORK_DIR)
+ .trackerDirPath(trackerDirPath)
+ .build();
final int expectedLines = 1;
int seenLines = 0;
List<Event> events = reader.readEvents(10);
@@ -434,18 +437,16 @@ public class TestReliableSpoolingFileEventReader {
Assert.assertEquals(expectedLines, seenLines);
}
- private void templateTestForLargeNumberOfFiles(ConsumeOrder order,
- Comparator<Long> comparator,
- int N) throws IOException {
+ private void templateTestForLargeNumberOfFiles(ConsumeOrder order, Comparator<Long> comparator,
+ int N) throws IOException {
File dir = null;
try {
- dir = new File(
- "target/test/work/" + this.getClass().getSimpleName() +
- "_large");
+ dir = new File("target/test/work/" + this.getClass().getSimpleName() + "_large");
Files.createParentDirs(new File(dir, "dummy"));
- ReliableEventReader reader
- = new ReliableSpoolingFileEventReader.Builder()
- .spoolDirectory(dir).consumeOrder(order).build();
+ ReliableEventReader reader =
+ new ReliableSpoolingFileEventReader.Builder().spoolDirectory(dir)
+ .consumeOrder(order)
+ .build();
Map<Long, List<String>> expected;
if (comparator == null) {
expected = new TreeMap<Long, List<String>>();
@@ -476,16 +477,14 @@ public class TestReliableSpoolingFileEventReader {
List<Event> events;
events = reader.readEvents(10);
for (Event e : events) {
- if (order == ConsumeOrder.RANDOM) {
+ if (order == ConsumeOrder.RANDOM) {
Assert.assertTrue(expectedList.remove(new String(e.getBody())));
} else {
- Assert.assertEquals(
- ((ArrayList<String>) expectedList).get(0),
- new String(e.getBody()));
+ Assert.assertEquals(((ArrayList<String>) expectedList).get(0), new String(e.getBody()));
((ArrayList<String>) expectedList).remove(0);
}
}
- reader.commit();
+ reader.commit();
}
} finally {
deleteDir(dir);
@@ -493,23 +492,24 @@ public class TestReliableSpoolingFileEventReader {
}
private void readEventsForFilesInDir(File dir, ReliableEventReader reader,
- Collection<String> actual) throws IOException {
+ Collection<String> actual) throws IOException {
readEventsForFilesInDir(dir, reader, actual, null, null);
}
/* Read events, one for each file in the given directory. */
- private void readEventsForFilesInDir(File dir, ReliableEventReader reader,
- Collection<String> actual, Semaphore semaphore1, Semaphore semaphore2) throws IOException {
+ private void readEventsForFilesInDir(File dir, ReliableEventReader reader,
+ Collection<String> actual, Semaphore semaphore1,
+ Semaphore semaphore2) throws IOException {
List<Event> events;
boolean executed = false;
- for (int i=0; i < listFiles(dir).size(); i++) {
+ for (int i = 0; i < listFiles(dir).size(); i++) {
events = reader.readEvents(10);
for (Event e : events) {
actual.add(new String(e.getBody()));
}
reader.commit();
try {
- if(!executed) {
+ if (!executed) {
executed = true;
if (semaphore1 != null) {
semaphore1.release();
@@ -533,8 +533,7 @@ public class TestReliableSpoolingFileEventReader {
}
private static List<File> listFiles(File dir) {
- List<File> files = Lists.newArrayList(dir.listFiles(new FileFilter
- () {
+ List<File> files = Lists.newArrayList(dir.listFiles(new FileFilter() {
@Override
public boolean accept(File pathname) {
return !pathname.isDirectory();
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/formatter/output/TestBucketPath.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/formatter/output/TestBucketPath.java b/flume-ng-core/src/test/java/org/apache/flume/formatter/output/TestBucketPath.java
index 21b972b..b1b828a 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/formatter/output/TestBucketPath.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/formatter/output/TestBucketPath.java
@@ -18,6 +18,12 @@
package org.apache.flume.formatter.output;
+import org.apache.flume.Clock;
+import org.joda.time.format.DateTimeFormatter;
+import org.joda.time.format.ISODateTimeFormat;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
import java.text.SimpleDateFormat;
import java.util.Calendar;
@@ -26,21 +32,15 @@ import java.util.HashMap;
import java.util.Map;
import java.util.TimeZone;
-import org.apache.flume.Clock;
-import org.joda.time.format.DateTimeFormatter;
-import org.joda.time.format.ISODateTimeFormat;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
public class TestBucketPath {
Calendar cal;
Map<String, String> headers;
+
@Before
- public void setUp(){
+ public void setUp() {
cal = Calendar.getInstance();
cal.set(2012, 5, 23, 13, 46, 33);
cal.set(Calendar.MILLISECOND, 234);
@@ -49,7 +49,7 @@ public class TestBucketPath {
}
@Test
- public void testDateFormatCache(){
+ public void testDateFormatCache() {
TimeZone utcTimeZone = TimeZone.getTimeZone("UTC");
String test = "%c";
BucketPath.escapeString(
@@ -60,7 +60,7 @@ public class TestBucketPath {
SimpleDateFormat format = new SimpleDateFormat("EEE MMM d HH:mm:ss yyyy");
Date d = new Date(cal.getTimeInMillis());
String expectedString = format.format(d);
- System.out.println("Expected String: "+ expectedString);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@@ -76,7 +76,7 @@ public class TestBucketPath {
SimpleDateFormat format = new SimpleDateFormat("EEE MMM d HH:mm:ss yyyy");
Date d = new Date(cal2.getTimeInMillis());
String expectedString = format.format(d);
- System.out.println("Expected String: "+ expectedString);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@@ -89,8 +89,8 @@ public class TestBucketPath {
Calendar cal2 = Calendar.getInstance();
cal2.set(2012, 5, 23, 13, 45, 0);
cal2.set(Calendar.MILLISECOND, 0);
- String expectedString = String.valueOf(cal2.getTimeInMillis()/1000);
- System.out.println("Expected String: "+ expectedString);
+ String expectedString = String.valueOf(cal2.getTimeInMillis() / 1000);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@@ -103,13 +103,13 @@ public class TestBucketPath {
Calendar cal2 = Calendar.getInstance();
cal2.set(2012, 5, 23, 13, 46, 30);
cal2.set(Calendar.MILLISECOND, 0);
- String expectedString = String.valueOf(cal2.getTimeInMillis()/1000);
- System.out.println("Expected String: "+ expectedString);
+ String expectedString = String.valueOf(cal2.getTimeInMillis() / 1000);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@Test
- public void testNoRounding(){
+ public void testNoRounding() {
String test = "%c";
String escapedString = BucketPath.escapeString(
test, headers, false, Calendar.HOUR_OF_DAY, 12);
@@ -117,19 +117,19 @@ public class TestBucketPath {
SimpleDateFormat format = new SimpleDateFormat("EEE MMM d HH:mm:ss yyyy");
Date d = new Date(cal.getTimeInMillis());
String expectedString = format.format(d);
- System.out.println("Expected String: "+ expectedString);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@Test
- public void testNoPadding(){
+ public void testNoPadding() {
Calendar calender;
Map<String, String> calender_timestamp;
calender = Calendar.getInstance();
-
+
//Check single digit dates
- calender.set(2014, (5-1), 3, 13, 46, 33);
+ calender.set(2014, (5 - 1), 3, 13, 46, 33);
calender_timestamp = new HashMap<String, String>();
calender_timestamp.put("timestamp", String.valueOf(calender.getTimeInMillis()));
SimpleDateFormat format = new SimpleDateFormat("M-d");
@@ -141,19 +141,19 @@ public class TestBucketPath {
String expectedString = format.format(d);
//Check two digit dates
- calender.set(2014, (11-1), 13, 13, 46, 33);
+ calender.set(2014, (11 - 1), 13, 13, 46, 33);
calender_timestamp.put("timestamp", String.valueOf(calender.getTimeInMillis()));
- escapedString += " " + BucketPath.escapeString(
+ escapedString += " " + BucketPath.escapeString(
test, calender_timestamp, false, Calendar.HOUR_OF_DAY, 12);
System.out.println("Escaped String: " + escapedString);
d = new Date(calender.getTimeInMillis());
- expectedString += " " + format.format(d);
- System.out.println("Expected String: "+ expectedString);
+ expectedString += " " + format.format(d);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
@Test
- public void testDateFormatTimeZone(){
+ public void testDateFormatTimeZone() {
TimeZone utcTimeZone = TimeZone.getTimeZone("UTC");
String test = "%c";
String escapedString = BucketPath.escapeString(
@@ -163,7 +163,7 @@ public class TestBucketPath {
format.setTimeZone(utcTimeZone);
Date d = new Date(cal.getTimeInMillis());
String expectedString = format.format(d);
- System.out.println("Expected String: "+ expectedString);
+ System.out.println("Expected String: " + expectedString);
Assert.assertEquals(expectedString, escapedString);
}
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/instrumentation/TestMonitoredCounterGroup.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/TestMonitoredCounterGroup.java b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/TestMonitoredCounterGroup.java
index b1f637f..7db535e 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/TestMonitoredCounterGroup.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/TestMonitoredCounterGroup.java
@@ -18,22 +18,15 @@
*/
package org.apache.flume.instrumentation;
-import java.lang.management.ManagementFactory;
-import java.util.Random;
-
-import javax.management.AttributeNotFoundException;
-import javax.management.InstanceNotFoundException;
-import javax.management.MBeanException;
-import javax.management.MBeanInfo;
-import javax.management.MBeanServer;
-import javax.management.ObjectName;
-import javax.management.ReflectionException;
-
import junit.framework.Assert;
-
import org.junit.Before;
import org.junit.Test;
+import javax.management.MBeanServer;
+import javax.management.ObjectName;
+import java.lang.management.ManagementFactory;
+import java.util.Random;
+
public class TestMonitoredCounterGroup {
private static final int MAX_BOUNDS = 1000;
@@ -61,7 +54,6 @@ public class TestMonitoredCounterGroup {
private static final String SRC_ATTR_APPEND_BATCH_ACCEPTED_COUNT =
"AppendBatchAcceptedCount";
-
private static final String CH_ATTR_CHANNEL_SIZE = "ChannelSize";
private static final String CH_ATTR_EVENT_PUT_ATTEMPT =
"EventPutAttemptCount";
@@ -122,28 +114,28 @@ public class TestMonitoredCounterGroup {
int eventDrainAttempt = random.nextInt(MAX_BOUNDS);
int eventDrainSuccess = random.nextInt(MAX_BOUNDS);
- for (int i = 0; i<connCreated; i++) {
+ for (int i = 0; i < connCreated; i++) {
skc.incrementConnectionCreatedCount();
}
- for (int i = 0; i<connClosed; i++) {
+ for (int i = 0; i < connClosed; i++) {
skc.incrementConnectionClosedCount();
}
- for (int i = 0; i<connFailed; i++) {
+ for (int i = 0; i < connFailed; i++) {
skc.incrementConnectionFailedCount();
}
- for (int i = 0; i<batchEmpty; i++) {
+ for (int i = 0; i < batchEmpty; i++) {
skc.incrementBatchEmptyCount();
}
- for (int i = 0; i<batchUnderflow; i++) {
+ for (int i = 0; i < batchUnderflow; i++) {
skc.incrementBatchUnderflowCount();
}
- for (int i = 0; i<batchComplete; i++) {
+ for (int i = 0; i < batchComplete; i++) {
skc.incrementBatchCompleteCount();
}
- for (int i = 0; i<eventDrainAttempt; i++) {
+ for (int i = 0; i < eventDrainAttempt; i++) {
skc.incrementEventDrainAttemptCount();
}
- for (int i = 0; i<eventDrainSuccess; i++) {
+ for (int i = 0; i < eventDrainSuccess; i++) {
skc.incrementEventDrainSuccessCount();
}
@@ -204,10 +196,10 @@ public class TestMonitoredCounterGroup {
int numEventTakeSuccess = random.nextInt(MAX_BOUNDS);
chc.setChannelSize(numChannelSize);
- for (int i = 0; i<numEventPutAttempt; i++) {
+ for (int i = 0; i < numEventPutAttempt; i++) {
chc.incrementEventPutAttemptCount();
}
- for (int i = 0; i<numEventTakeAttempt; i++) {
+ for (int i = 0; i < numEventTakeAttempt; i++) {
chc.incrementEventTakeAttemptCount();
}
chc.addToEventPutSuccessCount(numEventPutSuccess);
@@ -264,16 +256,16 @@ public class TestMonitoredCounterGroup {
srcc.addToEventReceivedCount(numEventReceived);
srcc.addToEventAcceptedCount(numEventAccepted);
- for (int i = 0; i<numAppendReceived; i++) {
+ for (int i = 0; i < numAppendReceived; i++) {
srcc.incrementAppendReceivedCount();
}
- for (int i = 0; i<numAppendAccepted; i++) {
+ for (int i = 0; i < numAppendAccepted; i++) {
srcc.incrementAppendAcceptedCount();
}
- for (int i = 0; i<numAppendBatchReceived; i++) {
+ for (int i = 0; i < numAppendBatchReceived; i++) {
srcc.incrementAppendBatchReceivedCount();
}
- for (int i = 0; i<numAppendBatchAccepted; i++) {
+ for (int i = 0; i < numAppendBatchAccepted; i++) {
srcc.incrementAppendBatchAcceptedCount();
}
@@ -302,11 +294,11 @@ public class TestMonitoredCounterGroup {
int numEventReceived2 = random.nextInt(MAX_BOUNDS);
int numEventAccepted2 = random.nextInt(MAX_BOUNDS);
- for (int i = 0; i<numEventReceived2; i++) {
+ for (int i = 0; i < numEventReceived2; i++) {
srcc.incrementEventReceivedCount();
}
- for (int i = 0; i<numEventAccepted2; i++) {
+ for (int i = 0; i < numEventAccepted2; i++) {
srcc.incrementEventAcceptedCount();
}
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/instrumentation/http/TestHTTPMetricsServer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/http/TestHTTPMetricsServer.java b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/http/TestHTTPMetricsServer.java
index eb2d02d..09d419f 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/http/TestHTTPMetricsServer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/http/TestHTTPMetricsServer.java
@@ -20,12 +20,6 @@ package org.apache.flume.instrumentation.http;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
-import java.io.BufferedReader;
-import java.io.InputStreamReader;
-import java.lang.reflect.Type;
-import java.net.HttpURLConnection;
-import java.net.URL;
-import java.util.Map;
import org.apache.flume.Channel;
import org.apache.flume.Context;
import org.apache.flume.Transaction;
@@ -39,6 +33,12 @@ import org.junit.Assert;
import org.junit.Test;
import javax.servlet.http.HttpServletResponse;
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.lang.reflect.Type;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.Map;
/**
*
@@ -47,9 +47,7 @@ public class TestHTTPMetricsServer {
Channel memChannel = new MemoryChannel();
Channel pmemChannel = new PseudoTxnMemoryChannel();
- Type mapType =
- new TypeToken<Map<String, Map<String, String>>>() {
- }.getType();
+ Type mapType = new TypeToken<Map<String, Map<String, String>>>() {}.getType();
Gson gson = new Gson();
@Test
@@ -99,7 +97,7 @@ public class TestHTTPMetricsServer {
private void testWithPort(int port) throws Exception {
MonitorService srv = new HTTPMetricsServer();
Context context = new Context();
- if(port > 1024){
+ if (port > 1024) {
context.put(HTTPMetricsServer.CONFIG_PORT, String.valueOf(port));
} else {
port = HTTPMetricsServer.DEFAULT_PORT;
@@ -139,8 +137,7 @@ public class TestHTTPMetricsServer {
doTestForbiddenMethods(4432,"OPTIONS");
}
- public void doTestForbiddenMethods(int port, String method)
- throws Exception {
+ public void doTestForbiddenMethods(int port, String method) throws Exception {
MonitorService srv = new HTTPMetricsServer();
Context context = new Context();
if (port > 1024) {
@@ -154,8 +151,7 @@ public class TestHTTPMetricsServer {
URL url = new URL("http://0.0.0.0:" + String.valueOf(port) + "/metrics");
HttpURLConnection conn = (HttpURLConnection) url.openConnection();
conn.setRequestMethod(method);
- Assert.assertEquals(HttpServletResponse.SC_FORBIDDEN,
- conn.getResponseCode());
+ Assert.assertEquals(HttpServletResponse.SC_FORBIDDEN, conn.getResponseCode());
srv.stop();
}
}
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/instrumentation/kafka/KafkaSourceCounterTest.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/kafka/KafkaSourceCounterTest.java b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/kafka/KafkaSourceCounterTest.java
index 4a71265..6d64c53 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/instrumentation/kafka/KafkaSourceCounterTest.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/instrumentation/kafka/KafkaSourceCounterTest.java
@@ -6,9 +6,9 @@
* 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
- *
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
* 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.
@@ -23,41 +23,41 @@ import org.junit.Test;
public class KafkaSourceCounterTest {
- KafkaSourceCounter counter;
-
- @Before
- public void setUp() throws Exception {
- counter = new KafkaSourceCounter("test");
- }
-
- @Test
- public void testAddToKafkaEventGetTimer() throws Exception {
- Assert.assertEquals(1L, counter.addToKafkaEventGetTimer(1L));
- }
-
- @Test
- public void testAddToKafkaCommitTimer() throws Exception {
- Assert.assertEquals(1L, counter.addToKafkaCommitTimer(1L));
- }
-
- @Test
- public void testIncrementKafkaEmptyCount() throws Exception {
- Assert.assertEquals(1L, counter.incrementKafkaEmptyCount());
- }
-
- @Test
- public void testGetKafkaCommitTimer() throws Exception {
- Assert.assertEquals(0, counter.getKafkaCommitTimer());
- }
-
- @Test
- public void testGetKafkaEventGetTimer() throws Exception {
- Assert.assertEquals(0, counter.getKafkaEventGetTimer());
- }
-
- @Test
- public void testGetKafkaEmptyCount() throws Exception {
- Assert.assertEquals(0, counter.getKafkaEmptyCount());
- }
+ KafkaSourceCounter counter;
+
+ @Before
+ public void setUp() throws Exception {
+ counter = new KafkaSourceCounter("test");
+ }
+
+ @Test
+ public void testAddToKafkaEventGetTimer() throws Exception {
+ Assert.assertEquals(1L, counter.addToKafkaEventGetTimer(1L));
+ }
+
+ @Test
+ public void testAddToKafkaCommitTimer() throws Exception {
+ Assert.assertEquals(1L, counter.addToKafkaCommitTimer(1L));
+ }
+
+ @Test
+ public void testIncrementKafkaEmptyCount() throws Exception {
+ Assert.assertEquals(1L, counter.incrementKafkaEmptyCount());
+ }
+
+ @Test
+ public void testGetKafkaCommitTimer() throws Exception {
+ Assert.assertEquals(0, counter.getKafkaCommitTimer());
+ }
+
+ @Test
+ public void testGetKafkaEventGetTimer() throws Exception {
+ Assert.assertEquals(0, counter.getKafkaEventGetTimer());
+ }
+
+ @Test
+ public void testGetKafkaEmptyCount() throws Exception {
+ Assert.assertEquals(0, counter.getKafkaEmptyCount());
+ }
}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorMillisSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorMillisSerializer.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorMillisSerializer.java
index ac46131..dd42079 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorMillisSerializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorMillisSerializer.java
@@ -18,7 +18,6 @@
package org.apache.flume.interceptor;
import junit.framework.Assert;
-
import org.apache.flume.Context;
import org.joda.time.format.DateTimeFormat;
import org.joda.time.format.DateTimeFormatter;
@@ -29,7 +28,8 @@ public class TestRegexExtractorInterceptorMillisSerializer {
@Test
public void shouldRequirePatternInConfiguration() {
try {
- RegexExtractorInterceptorMillisSerializer fixture = new RegexExtractorInterceptorMillisSerializer();
+ RegexExtractorInterceptorMillisSerializer fixture =
+ new RegexExtractorInterceptorMillisSerializer();
fixture.configure(new Context());
Assert.fail();
} catch (IllegalArgumentException ex) {
@@ -40,7 +40,8 @@ public class TestRegexExtractorInterceptorMillisSerializer {
@Test
public void shouldRequireValidPatternInConfiguration() {
try {
- RegexExtractorInterceptorMillisSerializer fixture = new RegexExtractorInterceptorMillisSerializer();
+ RegexExtractorInterceptorMillisSerializer fixture =
+ new RegexExtractorInterceptorMillisSerializer();
Context context = new Context();
context.put("pattern", "ABCDEFG");
fixture.configure(context);
@@ -52,7 +53,8 @@ public class TestRegexExtractorInterceptorMillisSerializer {
@Test
public void shouldReturnMillisFromPattern() {
- RegexExtractorInterceptorMillisSerializer fixture = new RegexExtractorInterceptorMillisSerializer();
+ RegexExtractorInterceptorMillisSerializer fixture =
+ new RegexExtractorInterceptorMillisSerializer();
Context context = new Context();
String pattern = "yyyy-MM-dd HH:mm:ss";
context.put("pattern", pattern);
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorPassThroughSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorPassThroughSerializer.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorPassThroughSerializer.java
index 569c274..33003e6 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorPassThroughSerializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestRegexExtractorInterceptorPassThroughSerializer.java
@@ -26,7 +26,8 @@ public class TestRegexExtractorInterceptorPassThroughSerializer {
@Test
public void shouldReturnSameValue() {
- RegexExtractorInterceptorPassThroughSerializer fixture = new RegexExtractorInterceptorPassThroughSerializer();
+ RegexExtractorInterceptorPassThroughSerializer fixture =
+ new RegexExtractorInterceptorPassThroughSerializer();
fixture.configure(new Context());
String input = "testing (1,2,3,4)";
Assert.assertEquals(input, fixture.serialize(input));
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
index 2ab15f5..616b86b 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/interceptor/TestSearchAndReplaceInterceptor.java
@@ -35,7 +35,7 @@ public class TestSearchAndReplaceInterceptor {
private void testSearchReplace(Context context, String input, String output)
throws Exception {
- Interceptor.Builder builder = InterceptorBuilderFactory.newInstance(
+ Interceptor.Builder builder = InterceptorBuilderFactory.newInstance(
InterceptorType.SEARCH_REPLACE.toString());
builder.configure(context);
Interceptor interceptor = builder.build();
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/SyslogAvroEventSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/SyslogAvroEventSerializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/SyslogAvroEventSerializer.java
index 896eced..05af3b1 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/SyslogAvroEventSerializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/SyslogAvroEventSerializer.java
@@ -231,20 +231,45 @@ public class SyslogAvroEventSerializer
private String hostname = "";
private String message = "";
- public void setFacility(int f) { facility = f; }
- public int getFacility() { return facility; }
+ public void setFacility(int f) {
+ facility = f;
+ }
+
+ public int getFacility() {
+ return facility;
+ }
+
+ public void setSeverity(int s) {
+ severity = s;
+ }
- public void setSeverity(int s) { severity = s; }
- public int getSeverity() { return severity; }
+ public int getSeverity() {
+ return severity;
+ }
- public void setTimestamp(long t) { timestamp = t; }
- public long getTimestamp() { return timestamp; }
+ public void setTimestamp(long t) {
+ timestamp = t;
+ }
- public void setHostname(String h) { hostname = h; }
- public String getHostname() { return hostname; }
+ public long getTimestamp() {
+ return timestamp;
+ }
- public void setMessage(String m) { message = m; }
- public String getMessage() { return message; }
+ public void setHostname(String h) {
+ hostname = h;
+ }
+
+ public String getHostname() {
+ return hostname;
+ }
+
+ public void setMessage(String m) {
+ message = m;
+ }
+
+ public String getMessage() {
+ return message;
+ }
@Override
public String toString() {
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/TestAvroEventDeserializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestAvroEventDeserializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestAvroEventDeserializer.java
index 6f9ddc2..b95433f 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestAvroEventDeserializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestAvroEventDeserializer.java
@@ -49,6 +49,7 @@ public class TestAvroEventDeserializer {
LoggerFactory.getLogger(TestAvroEventDeserializer.class);
private static final Schema schema;
+
static {
schema = Schema.createRecord("MyRecord", "", "org.apache.flume", false);
Schema.Field field = new Schema.Field("foo",
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
index e52affb..0c76cc9 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestDurablePositionTracker.java
@@ -31,8 +31,7 @@ import java.net.URL;
public class TestDurablePositionTracker {
- private static final Logger logger = LoggerFactory.getLogger
- (TestDurablePositionTracker.class);
+ private static final Logger logger = LoggerFactory.getLogger(TestDurablePositionTracker.class);
@Test
public void testBasicTracker() throws IOException {
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/TestFlumeEventAvroEventSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestFlumeEventAvroEventSerializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestFlumeEventAvroEventSerializer.java
index 3860b5e..ded3b13 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestFlumeEventAvroEventSerializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestFlumeEventAvroEventSerializer.java
@@ -18,14 +18,7 @@
*/
package org.apache.flume.serialization;
-import java.io.File;
-import java.io.FileNotFoundException;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.io.OutputStream;
-import java.nio.ByteBuffer;
-import java.nio.charset.CharsetDecoder;
-
+import com.google.common.base.Charsets;
import org.apache.avro.file.DataFileReader;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericDatumReader;
@@ -38,7 +31,13 @@ import org.junit.Assert;
import org.junit.Assume;
import org.junit.Test;
-import com.google.common.base.Charsets;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharsetDecoder;
public class TestFlumeEventAvroEventSerializer {
@@ -77,17 +76,16 @@ public class TestFlumeEventAvroEventSerializer {
throws FileNotFoundException, IOException {
// Snappy currently broken on Mac in OpenJDK 7 per FLUME-2012
Assume.assumeTrue(!"Mac OS X".equals(System.getProperty("os.name")) ||
- !System.getProperty("java.version").startsWith("1.7."));
+ !System.getProperty("java.version").startsWith("1.7."));
createAvroFile(TESTFILE, "snappy");
validateAvroFile(TESTFILE);
FileUtils.forceDelete(TESTFILE);
}
- public void createAvroFile(File file, String codec)
- throws FileNotFoundException, IOException {
+ public void createAvroFile(File file, String codec) throws FileNotFoundException, IOException {
- if(file.exists()){
+ if (file.exists()) {
FileUtils.forceDelete(file);
}
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
index 631bdfe..9f336eb 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestResettableFileInputStream.java
@@ -27,8 +27,6 @@ import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.junit.Assert.*;
-
import java.io.BufferedOutputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
@@ -39,15 +37,21 @@ import java.nio.charset.Charset;
import java.nio.charset.MalformedInputException;
import java.util.List;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
public class TestResettableFileInputStream {
private static final boolean CLEANUP = true;
private static final File WORK_DIR =
new File("target/test/work").getAbsoluteFile();
- private static final Logger logger = LoggerFactory.getLogger
- (TestResettableFileInputStream.class);
+ private static final Logger logger =
+ LoggerFactory.getLogger(TestResettableFileInputStream.class);
- private File file, meta;
+ private File file;
+ private File meta;
@Before
public void setup() throws Exception {
@@ -156,7 +160,8 @@ public class TestResettableFileInputStream {
public void testUtf16BOMAndSurrogatePairRead() throws IOException {
ByteArrayOutputStream out = new ByteArrayOutputStream();
generateUtf16SurrogatePairSequence(out);
- // buffer now contains 1 BOM and 2 chars (1 surrogate pair) and 6 bytes total (including 2-byte BOM)
+ // buffer now contains 1 BOM and 2 chars (1 surrogate pair) and 6 bytes total
+ // (including 2-byte BOM)
Files.write(out.toByteArray(), file);
ResettableInputStream in = initInputStream(8, Charsets.UTF_16, DecodeErrorPolicy.FAIL);
String result = readLine(in, 2);
@@ -176,7 +181,8 @@ public class TestResettableFileInputStream {
generateShiftJis2ByteSequence(out);
// buffer now contains 8 chars and 10 bytes total
Files.write(out.toByteArray(), file);
- ResettableInputStream in = initInputStream(8, Charset.forName("Shift_JIS"), DecodeErrorPolicy.FAIL);
+ ResettableInputStream in = initInputStream(8, Charset.forName("Shift_JIS"),
+ DecodeErrorPolicy.FAIL);
String result = readLine(in, 8);
assertEquals("1234567\u4E9C\n", result);
}
@@ -215,7 +221,7 @@ public class TestResettableFileInputStream {
String javaVersionStr = System.getProperty("java.version");
double javaVersion = Double.parseDouble(javaVersionStr.substring(0, 3));
- if(javaVersion < 1.8) {
+ if (javaVersion < 1.8) {
assertTrue(preJdk8ExpectedStr.replaceAll("X", "\ufffd").equals(sb.toString()));
} else {
assertTrue(expectedStr.replaceAll("X", "\ufffd").equals(sb.toString()));
@@ -508,8 +514,8 @@ public class TestResettableFileInputStream {
return initInputStream(2048, Charsets.UTF_8, policy);
}
- private ResettableInputStream initInputStream(int bufferSize, Charset charset, DecodeErrorPolicy policy)
- throws IOException {
+ private ResettableInputStream initInputStream(int bufferSize, Charset charset,
+ DecodeErrorPolicy policy) throws IOException {
PositionTracker tracker = new DurablePositionTracker(meta, file.getPath());
ResettableInputStream in = new ResettableFileInputStream(file, tracker,
bufferSize, charset, policy);
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/serialization/TestSyslogAvroEventSerializer.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestSyslogAvroEventSerializer.java b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestSyslogAvroEventSerializer.java
index 7bd342a..ba9f4ab 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/serialization/TestSyslogAvroEventSerializer.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/serialization/TestSyslogAvroEventSerializer.java
@@ -75,7 +75,7 @@ public class TestSyslogAvroEventSerializer {
public void test() throws FileNotFoundException, IOException {
// Snappy currently broken on Mac in OpenJDK 7 per FLUME-2012
Assume.assumeTrue(!"Mac OS X".equals(System.getProperty("os.name")) ||
- !System.getProperty("java.version").startsWith("1.7."));
+ !System.getProperty("java.version").startsWith("1.7."));
//Schema schema = new Schema.Parser().parse(schemaFile);
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/sink/TestAvroSink.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestAvroSink.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestAvroSink.java
index 757a536..0f3f9ec 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestAvroSink.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestAvroSink.java
@@ -19,28 +19,10 @@
package org.apache.flume.sink;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.nio.ByteBuffer;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.concurrent.atomic.AtomicLong;
-
import com.google.common.base.Charsets;
-import java.io.FileInputStream;
-import java.security.KeyStore;
-import java.security.Security;
-import java.util.concurrent.Executors;
-import javax.net.ssl.KeyManagerFactory;
-import javax.net.ssl.SSLContext;
-import javax.net.ssl.SSLEngine;
import org.apache.avro.AvroRemoteException;
import org.apache.avro.ipc.NettyServer;
-import org.apache.avro.ipc.NettyTransceiver;
import org.apache.avro.ipc.Server;
-import org.apache.avro.ipc.specific.SpecificRequestor;
import org.apache.avro.ipc.specific.SpecificResponder;
import org.apache.flume.Channel;
import org.apache.flume.ChannelSelector;
@@ -49,8 +31,8 @@ import org.apache.flume.Event;
import org.apache.flume.EventDeliveryException;
import org.apache.flume.Sink;
import org.apache.flume.Transaction;
-import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.api.RpcClient;
+import org.apache.flume.channel.ChannelProcessor;
import org.apache.flume.channel.MemoryChannel;
import org.apache.flume.channel.ReplicatingChannelSelector;
import org.apache.flume.conf.Configurables;
@@ -61,19 +43,30 @@ import org.apache.flume.source.AvroSource;
import org.apache.flume.source.avro.AvroFlumeEvent;
import org.apache.flume.source.avro.AvroSourceProtocol;
import org.apache.flume.source.avro.Status;
-import org.jboss.netty.channel.ChannelException;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;
import org.jboss.netty.handler.ssl.SslHandler;
import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Ignore;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLEngine;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.nio.charset.Charset;
+import java.security.KeyStore;
+import java.security.Security;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicLong;
+
public class TestAvroSink {
private static final Logger logger = LoggerFactory
@@ -90,7 +83,9 @@ public class TestAvroSink {
}
public void setUp(String compressionType, int compressionLevel) {
- if (sink != null) { throw new RuntimeException("double setup");}
+ if (sink != null) {
+ throw new RuntimeException("double setup");
+ }
sink = new AvroSink();
channel = new MemoryChannel();
@@ -607,8 +602,9 @@ public class TestAvroSink {
}
@Test
- public void testSslSinkWithNonTrustedCert() throws InterruptedException,
- EventDeliveryException, InstantiationException, IllegalAccessException {
+ public void testSslSinkWithNonTrustedCert()
+ throws InterruptedException, EventDeliveryException, InstantiationException,
+ IllegalAccessException {
setUp();
Event event = EventBuilder.withBody("test event 1", Charsets.UTF_8);
Server server = createSslServer(new MockAvroServer());
@@ -662,37 +658,38 @@ public class TestAvroSink {
}
@Test
- public void testRequestWithNoCompression() throws InterruptedException, IOException, EventDeliveryException {
-
+ public void testRequestWithNoCompression()
+ throws InterruptedException, IOException, EventDeliveryException {
doRequest(false, false, 6);
}
@Test
- public void testRequestWithCompressionOnClientAndServerOnLevel0() throws InterruptedException, IOException, EventDeliveryException {
-
+ public void testRequestWithCompressionOnClientAndServerOnLevel0()
+ throws InterruptedException, IOException, EventDeliveryException {
doRequest(true, true, 0);
}
@Test
- public void testRequestWithCompressionOnClientAndServerOnLevel1() throws InterruptedException, IOException, EventDeliveryException {
-
+ public void testRequestWithCompressionOnClientAndServerOnLevel1()
+ throws InterruptedException, IOException, EventDeliveryException {
doRequest(true, true, 1);
}
@Test
- public void testRequestWithCompressionOnClientAndServerOnLevel6() throws InterruptedException, IOException, EventDeliveryException {
-
+ public void testRequestWithCompressionOnClientAndServerOnLevel6()
+ throws InterruptedException, IOException, EventDeliveryException {
doRequest(true, true, 6);
}
@Test
- public void testRequestWithCompressionOnClientAndServerOnLevel9() throws InterruptedException, IOException, EventDeliveryException {
-
+ public void testRequestWithCompressionOnClientAndServerOnLevel9()
+ throws InterruptedException, IOException, EventDeliveryException {
doRequest(true, true, 9);
}
- private void doRequest(boolean serverEnableCompression, boolean clientEnableCompression, int compressionLevel) throws InterruptedException, IOException, EventDeliveryException {
-
+ private void doRequest(boolean serverEnableCompression, boolean clientEnableCompression,
+ int compressionLevel)
+ throws InterruptedException, IOException, EventDeliveryException {
if (clientEnableCompression) {
setUp("deflate", compressionLevel);
} else {
@@ -732,15 +729,12 @@ public class TestAvroSink {
source.start();
- Assert
- .assertTrue("Reached start or error", LifecycleController.waitForOneOf(
- source, LifecycleState.START_OR_ERROR));
- Assert.assertEquals("Server is started", LifecycleState.START,
- source.getLifecycleState());
-
+ Assert.assertTrue("Reached start or error",
+ LifecycleController.waitForOneOf(source, LifecycleState.START_OR_ERROR));
+ Assert.assertEquals("Server is started",
+ LifecycleState.START, source.getLifecycleState());
- Event event = EventBuilder.withBody("Hello avro",
- Charset.forName("UTF8"));
+ Event event = EventBuilder.withBody("Hello avro", Charset.forName("UTF8"));
sink.start();
@@ -858,7 +852,8 @@ public class TestAvroSink {
public SSLChannelPipelineFactory() {
}
- public SSLChannelPipelineFactory(String keystore, String keystorePassword, String keystoreType) {
+ public SSLChannelPipelineFactory(String keystore, String keystorePassword,
+ String keystoreType) {
this.keystore = keystore;
this.keystorePassword = keystorePassword;
this.keystoreType = keystoreType;
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
index 835f541..cf6cbbc 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestDefaultSinkFactory.java
@@ -37,7 +37,6 @@ public class TestDefaultSinkFactory {
@Test
public void testDuplicateCreate() {
-
Sink avroSink1 = sinkFactory.create("avroSink1", "avro");
Sink avroSink2 = sinkFactory.create("avroSink2", "avro");
@@ -55,7 +54,7 @@ public class TestDefaultSinkFactory {
}
private void verifySinkCreation(String name, String type, Class<?> typeClass)
- throws Exception {
+ throws Exception {
Sink sink = sinkFactory.create(name, type);
Assert.assertNotNull(sink);
Assert.assertTrue(typeClass.isInstance(sink));
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
index 3358cf4..8882056 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestFailoverSinkProcessor.java
@@ -94,7 +94,7 @@ public class TestFailoverSinkProcessor {
@Override
public Status process() throws EventDeliveryException {
- synchronized(this) {
+ synchronized (this) {
if (remaining <= 0) {
throw new EventDeliveryException("can't consume more");
}
@@ -107,7 +107,7 @@ public class TestFailoverSinkProcessor {
tx.close();
if (e != null) {
- synchronized(this) {
+ synchronized (this) {
remaining--;
}
written++;
@@ -167,7 +167,7 @@ public class TestFailoverSinkProcessor {
Assert.assertEquals(LifecycleState.START, s1.getLifecycleState());
Assert.assertEquals(LifecycleState.START, s2.getLifecycleState());
Assert.assertEquals(LifecycleState.START, s3.getLifecycleState());
- for(int i = 0; i < 15; i++) {
+ for (int i = 0; i < 15; i++) {
Transaction tx = ch.getTransaction();
tx.begin();
ch.put(EventBuilder.withBody("test".getBytes()));
@@ -178,7 +178,7 @@ public class TestFailoverSinkProcessor {
Assert.assertEquals(new Integer(10), s1.getWritten());
Assert.assertEquals(new Integer(5), s2.getWritten());
- for(int i = 0; i < 50; i++) {
+ for (int i = 0; i < 50; i++) {
Transaction tx = ch.getTransaction();
tx.begin();
ch.put(EventBuilder.withBody("test".getBytes()));
@@ -195,7 +195,7 @@ public class TestFailoverSinkProcessor {
// get us past the retry time for the failed sink
Thread.sleep(5000);
- for(int i = 0; i < 100; i++) {
+ for (int i = 0; i < 100; i++) {
Transaction tx = ch.getTransaction();
tx.begin();
ch.put(EventBuilder.withBody("test".getBytes()));
http://git-wip-us.apache.org/repos/asf/flume/blob/cfbf1156/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
----------------------------------------------------------------------
diff --git a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
index 7d95655..011d2d1 100644
--- a/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
+++ b/flume-ng-core/src/test/java/org/apache/flume/sink/TestLoadBalancingSinkProcessor.java
@@ -17,16 +17,7 @@
*/
package org.apache.flume.sink;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
import junit.framework.Assert;
-
import org.apache.flume.Channel;
import org.apache.flume.ChannelException;
import org.apache.flume.Context;
@@ -38,6 +29,14 @@ import org.apache.flume.Transaction;
import org.apache.flume.channel.AbstractChannel;
import org.junit.Test;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
public class TestLoadBalancingSinkProcessor {
private Context getContext(String selectorType, boolean backoff) {
@@ -62,8 +61,7 @@ public class TestLoadBalancingSinkProcessor {
return getProcessor(sinks, getContext(selectorType, backoff));
}
- private LoadBalancingSinkProcessor getProcessor(List<Sink> sinks, Context ctx)
- {
+ private LoadBalancingSinkProcessor getProcessor(List<Sink> sinks, Context ctx) {
LoadBalancingSinkProcessor lbsp = new LoadBalancingSinkProcessor();
lbsp.setSinks(sinks);
lbsp.configure(ctx);
@@ -77,7 +75,7 @@ public class TestLoadBalancingSinkProcessor {
// If no selector is specified, the round-robin selector should be used
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -185,7 +183,7 @@ public class TestLoadBalancingSinkProcessor {
// TODO: there is a remote possibility that s0 or s2
// never get hit by the random assignment
// and thus not backoffed, causing the test to fail
- for(int i=0; i < 50; i++) {
+ for (int i = 0; i < 50; i++) {
// a well behaved runner would always check the return.
lbsp.process();
}
@@ -214,7 +212,7 @@ public class TestLoadBalancingSinkProcessor {
public void testRandomPersistentFailure() throws Exception {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -244,7 +242,7 @@ public class TestLoadBalancingSinkProcessor {
}
Assert.assertTrue(s2.getEvents().size() == 0);
- Assert.assertTrue(s1.getEvents().size() + s3.getEvents().size() == 3*n);
+ Assert.assertTrue(s1.getEvents().size() + s3.getEvents().size() == 3 * n);
}
@Test
@@ -325,8 +323,6 @@ public class TestLoadBalancingSinkProcessor {
Assert.assertTrue("Miraculous distribution", sizeSet.size() > 1);
}
-
-
@Test
public void testRoundRobinOneActiveSink() throws Exception {
Channel ch = new MockChannel();
@@ -373,7 +369,7 @@ public class TestLoadBalancingSinkProcessor {
public void testRoundRobinPersistentFailure() throws Exception {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -404,7 +400,7 @@ public class TestLoadBalancingSinkProcessor {
Assert.assertTrue(s1.getEvents().size() == n);
Assert.assertTrue(s2.getEvents().size() == 0);
- Assert.assertTrue(s3.getEvents().size() == 2*n);
+ Assert.assertTrue(s3.getEvents().size() == 2 * n);
}
// test that even if the sink recovers immediately that it is kept out of commission briefly
@@ -413,7 +409,7 @@ public class TestLoadBalancingSinkProcessor {
public void testRoundRobinBackoffInitialFailure() throws EventDeliveryException {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -424,7 +420,7 @@ public class TestLoadBalancingSinkProcessor {
MockSink s2 = new MockSink(2);
s2.setChannel(ch);
- MockSink s3 = new MockSink(3);
+ MockSink s3 = new MockSink(3);
s3.setChannel(ch);
List<Sink> sinks = new ArrayList<Sink>();
@@ -449,14 +445,15 @@ public class TestLoadBalancingSinkProcessor {
Assert.assertEquals((3 * n) / 2, s1.getEvents().size());
Assert.assertEquals(1, s2.getEvents().size());
- Assert.assertEquals((3 * n) /2 - 1, s3.getEvents().size());
+ Assert.assertEquals((3 * n) / 2 - 1, s3.getEvents().size());
}
@Test
- public void testRoundRobinBackoffIncreasingBackoffs() throws EventDeliveryException, InterruptedException {
+ public void testRoundRobinBackoffIncreasingBackoffs()
+ throws EventDeliveryException, InterruptedException {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -468,7 +465,7 @@ public class TestLoadBalancingSinkProcessor {
s2.setChannel(ch);
s2.setFail(true);
- MockSink s3 = new MockSink(3);
+ MockSink s3 = new MockSink(3);
s3.setChannel(ch);
List<Sink> sinks = new ArrayList<Sink>();
@@ -508,10 +505,11 @@ public class TestLoadBalancingSinkProcessor {
}
@Test
- public void testRoundRobinBackoffFailureRecovery() throws EventDeliveryException, InterruptedException {
+ public void testRoundRobinBackoffFailureRecovery()
+ throws EventDeliveryException, InterruptedException {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -523,7 +521,7 @@ public class TestLoadBalancingSinkProcessor {
s2.setChannel(ch);
s2.setFail(true);
- MockSink s3 = new MockSink(3);
+ MockSink s3 = new MockSink(3);
s3.setChannel(ch);
List<Sink> sinks = new ArrayList<Sink>();
@@ -548,13 +546,12 @@ public class TestLoadBalancingSinkProcessor {
Assert.assertEquals(n, s3.getEvents().size());
}
-
@Test
public void testRoundRobinNoFailure() throws Exception {
Channel ch = new MockChannel();
int n = 100;
- int numEvents = 3*n;
+ int numEvents = 3 * n;
for (int i = 0; i < numEvents; i++) {
ch.put(new MockEvent("test" + i));
}
@@ -656,8 +653,9 @@ public class TestLoadBalancingSinkProcessor {
throw new EventDeliveryException("failed");
}
Event e = this.getChannel().take();
- if (e == null)
+ if (e == null) {
return Status.BACKOFF;
+ }
events.add(e);
return Status.READY;