SAMZA-412; replace assert calls in tests with appropriate JUnit asser…

…t methods

samza-api/src/test/java/org/apache/samza/config/TestConfig.java

@@ -19,13 +19,18 @@
 
 package org.apache.samza.config;
 
-import org.junit.Assert.* ;
-import org.junit.Test ;
-import java.util.Map ;
-import java.util.HashMap ;
+import static org.junit.Assert.assertEquals;
+
+import java.util.Map;
+import java.util.HashMap;
+
+import org.junit.Test;
 
 public class TestConfig {
-  // Utility methods to make it easier to tell the class of a primitive via overloaded args
+  /**
+   * Utility methods to make it easier to tell the class of a primitive via
+   * overloaded args
+   */
   Class getClass(long l) {
     return Long.class ;
   }
@@ -35,25 +40,25 @@ Class getClass(short s) {
   }
 
   @Test
-  public void testgetShortAndLong(){
-    Map<String, String> m = new HashMap<String, String>() { {
-      put("testkey", "11") ;
-    } } ;
+  public void testgetShortAndLong() {
+    Map<String, String> m = new HashMap<String, String>() {{
+      put("testkey", "11");
+    }};
 
-    MapConfig mc = new MapConfig(m) ;
-    short defaultShort=0 ;
-    long defaultLong=0 ;
+    MapConfig mc = new MapConfig(m);
+    short defaultShort = 0;
+    long defaultLong = 0;
 
-    Class c1 = getClass(mc.getShort("testkey")) ;
-    assert(c1 == Short.class) ;
+    Class c1 = getClass(mc.getShort("testkey"));
+    assertEquals(Short.class, c1);
 
-    Class c2 = getClass(mc.getShort("testkey", defaultShort)) ;
-    assert(c2 == Short.class) ;
+    Class c2 = getClass(mc.getShort("testkey", defaultShort));
+    assertEquals(Short.class, c2);
 
-    Class c3 = getClass(mc.getLong("testkey")) ;
-    assert(c3 == Long.class) ;
+    Class c3 = getClass(mc.getLong("testkey"));
+    assertEquals(Long.class, c3);
 
-    Class c4 = getClass(mc.getLong("testkey", defaultLong)) ;
-    assert(c4 == Long.class) ;
+    Class c4 = getClass(mc.getLong("testkey", defaultLong));
+    assertEquals(Long.class, c4);
   }
 }

samza-api/src/test/java/org/apache/samza/metrics/TestSlidingTimeWindowReservoir.java

@@ -19,22 +19,22 @@
 
 package org.apache.samza.metrics;
 
+import static org.mockito.Mockito.*;
 import static org.junit.Assert.*;
 
 import java.util.Arrays;
 
 import org.apache.samza.util.Clock;
 import org.junit.Test;
 
-import static org.mockito.Mockito.*;
-
 public class TestSlidingTimeWindowReservoir {
 
   private final Clock clock = mock(Clock.class);
 
   @Test
   public void testUpdateSizeSnapshot() {
-    SlidingTimeWindowReservoir slidingTimeWindowReservoir = new SlidingTimeWindowReservoir(300, clock);
+    SlidingTimeWindowReservoir slidingTimeWindowReservoir =
+        new SlidingTimeWindowReservoir(300, clock);
 
     when(clock.currentTimeMillis()).thenReturn(0L);
     slidingTimeWindowReservoir.update(1L);
@@ -49,24 +49,26 @@ public void testUpdateSizeSnapshot() {
 
     Snapshot snapshot = slidingTimeWindowReservoir.getSnapshot();
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(1L, 2L, 3L)));
-    assertTrue(snapshot.getSize() == 3);
+    assertEquals(3, snapshot.getSize());
   }
 
   @Test
   public void testDuplicateTime() {
-    SlidingTimeWindowReservoir slidingTimeWindowReservoir = new SlidingTimeWindowReservoir(300, clock);
+    SlidingTimeWindowReservoir slidingTimeWindowReservoir =
+        new SlidingTimeWindowReservoir(300, clock);
     when(clock.currentTimeMillis()).thenReturn(0L);
     slidingTimeWindowReservoir.update(1L);
     slidingTimeWindowReservoir.update(2L);
 
     Snapshot snapshot = slidingTimeWindowReservoir.getSnapshot();
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(1L, 2L)));
-    assertTrue(snapshot.getSize() == 2);
+    assertEquals(2, snapshot.getSize());
   }
 
   @Test
   public void testRemoveExpiredValues() {
-    SlidingTimeWindowReservoir slidingTimeWindowReservoir = new SlidingTimeWindowReservoir(300, clock);
+    SlidingTimeWindowReservoir slidingTimeWindowReservoir =
+        new SlidingTimeWindowReservoir(300, clock);
     when(clock.currentTimeMillis()).thenReturn(0L);
     slidingTimeWindowReservoir.update(1L);
 
@@ -81,6 +83,6 @@ public void testRemoveExpiredValues() {
 
     Snapshot snapshot = slidingTimeWindowReservoir.getSnapshot();
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(3L, 4L)));
-    assertTrue(snapshot.getSize() == 2);
+    assertEquals(2, snapshot.getSize());
   }
-}
+}

samza-api/src/test/java/org/apache/samza/metrics/TestTimer.java

@@ -28,7 +28,9 @@
 
 public class TestTimer {
 
-  // mock clock
+  /*
+   * Mock clock
+   */
   private final Clock clock = new Clock() {
     long value = 0;
 
@@ -46,7 +48,7 @@ public void testDefaultTimerUpdateAndGetSnapshot() {
 
     Snapshot snapshot = timer.getSnapshot();
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(1L, 2L)));
-    assertTrue(snapshot.getValues().size() == 2);
+    assertEquals(2, snapshot.getValues().size());
   }
 
   @Test
@@ -58,13 +60,13 @@ public void testTimerWithDifferentWindowSize() {
 
     Snapshot snapshot = timer.getSnapshot();
     assertTrue(snapshot.getValues().containsAll(Arrays.asList(1L, 2L, 3L)));
-    assertTrue(snapshot.getValues().size() == 3);
+    assertEquals(3, snapshot.getValues().size());
 
-    // the time is 500 for update(4L) because getSnapshot calls clock once + 3
+    // The time is 500 for update(4L) because getSnapshot calls clock once + 3
     // updates that call clock 3 times
     timer.update(4L);
     Snapshot snapshot2 = timer.getSnapshot();
     assertTrue(snapshot2.getValues().containsAll(Arrays.asList(3L, 4L)));
-    assertTrue(snapshot2.getValues().size() == 2);
+    assertEquals(2, snapshot2.getValues().size());
   }
-}
+}

samza-api/src/test/java/org/apache/samza/util/TestBlockingEnvelopeMap.java

@@ -22,6 +22,7 @@
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
+
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -30,6 +31,7 @@
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.samza.Partition;
 import org.apache.samza.system.IncomingMessageEnvelope;
 import org.apache.samza.system.SystemStreamPartition;

samza-api/src/test/java/org/apache/samza/util/TestNoOpMetricsRegistry.java

@@ -33,32 +33,38 @@ public void testNoOpMetricsHappyPath() {
     Counter counter1 = registry.newCounter("testc", "a");
     Counter counter2 = registry.newCounter("testc", "b");
     Counter counter3 = registry.newCounter("testc2", "c");
+
     Gauge<String> gauge1 = registry.newGauge("testg", "a", "1");
     Gauge<String> gauge2 = registry.newGauge("testg", "b", "2");
     Gauge<String> gauge3 = registry.newGauge("testg", "c", "3");
     Gauge<String> gauge4 = registry.newGauge("testg2", "d", "4");
+
     Timer timer1 = registry.newTimer("testt", "a");
     Timer timer2 = registry.newTimer("testt", "b");
     Timer timer3 = registry.newTimer("testt2", "c");
+
     counter1.inc();
     counter2.inc(2);
     counter3.inc(4);
+
     gauge1.set("5");
     gauge2.set("6");
     gauge3.set("7");
     gauge4.set("8");
+
     timer1.update(1L);
     timer2.update(2L);
     timer3.update(3L);
-    assertEquals(counter1.getCount(), 1);
-    assertEquals(counter2.getCount(), 2);
-    assertEquals(counter3.getCount(), 4);
-    assertEquals(gauge1.getValue(), "5");
-    assertEquals(gauge2.getValue(), "6");
-    assertEquals(gauge3.getValue(), "7");
-    assertEquals(gauge4.getValue(), "8");
-    assertEquals(timer1.getSnapshot().getAverage(), 1, 0);
-    assertEquals(timer2.getSnapshot().getAverage(), 2, 0);
-    assertEquals(timer3.getSnapshot().getAverage(), 3, 0);
+
+    assertEquals(1, counter1.getCount());
+    assertEquals(2, counter2.getCount());
+    assertEquals(4, counter3.getCount());
+    assertEquals("5", gauge1.getValue());
+    assertEquals("6", gauge2.getValue());
+    assertEquals("7", gauge3.getValue());
+    assertEquals("8", gauge4.getValue());
+    assertEquals(1, timer1.getSnapshot().getAverage(), 0);
+    assertEquals(2, timer2.getSnapshot().getAverage(), 0);
+    assertEquals(3, timer3.getSnapshot().getAverage(), 0);
   }
 }

samza-api/src/test/java/org/apache/samza/util/TestSinglePartitionWithoutOffsetsSystemAdmin.java

@@ -20,10 +20,12 @@
 package org.apache.samza.util;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
 
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+
 import org.apache.samza.Partition;
 import org.apache.samza.system.SystemStreamPartition;
 import org.apache.samza.system.SystemStreamMetadata;
@@ -36,12 +38,14 @@ public void testShouldGetASinglePartition() {
     Set<String> streamNames = new HashSet<String>();
     streamNames.add("a");
     streamNames.add("b");
+
     Map<String, SystemStreamMetadata> metadata = admin.getSystemStreamMetadata(streamNames);
-    assertEquals(metadata.size(), 2);
+    assertEquals(2, metadata.size());
     SystemStreamMetadata metadata1 = metadata.get("a");
     SystemStreamMetadata metadata2 = metadata.get("b");
+
     assertEquals(1, metadata1.getSystemStreamPartitionMetadata().size());
     assertEquals(1, metadata2.getSystemStreamPartitionMetadata().size());
-    assertEquals(null, metadata.get(new SystemStreamPartition("test-system", "c", new Partition(0))));
+    assertNull(metadata.get(new SystemStreamPartition("test-system", "c", new Partition(0))));
   }
-}
+}

samza-core/src/test/scala/org/apache/samza/checkpoint/TestCheckpointTool.scala

@@ -20,6 +20,7 @@
 package org.apache.samza.checkpoint
 
 import org.apache.samza.Partition
+import org.apache.samza.container.TaskName
 import org.apache.samza.checkpoint.TestCheckpointTool.{MockCheckpointManagerFactory, MockSystemFactory}
 import org.apache.samza.config.{Config, MapConfig, SystemConfig, TaskConfig}
 import org.apache.samza.metrics.MetricsRegistry
@@ -30,8 +31,8 @@ import org.mockito.Matchers._
 import org.mockito.Mockito._
 import org.scalatest.junit.AssertionsForJUnit
 import org.scalatest.mock.MockitoSugar
+
 import scala.collection.JavaConversions._
-import org.apache.samza.container.TaskName
 
 object TestCheckpointTool {
   var checkpointManager: CheckpointManager = null

samza-core/src/test/scala/org/apache/samza/checkpoint/TestOffsetManager.scala

@@ -19,7 +19,9 @@
 
 package org.apache.samza.checkpoint
 
-import scala.collection.JavaConversions._
+import java.util
+
+import org.apache.samza.container.TaskName
 import org.apache.samza.Partition
 import org.apache.samza.system.SystemStream
 import org.apache.samza.system.SystemStreamMetadata
@@ -30,10 +32,10 @@ import org.junit.{Ignore, Test}
 import org.apache.samza.SamzaException
 import org.apache.samza.config.MapConfig
 import org.apache.samza.system.SystemAdmin
-import java.util
-import org.apache.samza.container.TaskName
 import org.scalatest.Assertions.intercept
 
+import scala.collection.JavaConversions._
+
 class TestOffsetManager {
   @Test
   def testSystemShouldUseDefaults {
@@ -47,7 +49,7 @@ class TestOffsetManager {
     val offsetManager = OffsetManager(systemStreamMetadata, config)
     offsetManager.register(taskName, Set(systemStreamPartition))
     offsetManager.start
-    assertTrue(!offsetManager.getLastProcessedOffset(systemStreamPartition).isDefined)
+    assertFalse(offsetManager.getLastProcessedOffset(systemStreamPartition).isDefined)
     assertTrue(offsetManager.getStartingOffset(systemStreamPartition).isDefined)
     assertEquals("0", offsetManager.getStartingOffset(systemStreamPartition).get)
   }
@@ -232,7 +234,6 @@ class TestOffsetManager {
       override def writeChangeLogPartitionMapping(mapping: util.Map[TaskName, java.lang.Integer]): Unit = taskNameToPartitionMapping = mapping
 
       override def readChangeLogPartitionMapping(): util.Map[TaskName, java.lang.Integer] = taskNameToPartitionMapping
-
     }
   }
 

samza-core/src/test/scala/org/apache/samza/config/factories/TestPropertiesConfigFactory.scala

@@ -18,8 +18,10 @@
  */
 
 package org.apache.samza.config.factories
+
 import java.net.URI
 import java.io.File
+
 import org.apache.samza.SamzaException
 import org.junit.Assert._
 import org.junit.Test
@@ -30,7 +32,7 @@ class TestPropertiesConfigFactory {
   @Test
   def testCanReadPropertiesConfigFiles {
     val config = factory.getConfig(URI.create("file://%s/src/test/resources/test.properties" format new File(".").getCanonicalPath))
-    assert("bar".equals(config.get("foo")))
+    assertEquals("bar", config.get("foo"))
   }
 
   @Test

samza-core/src/test/scala/org/apache/samza/container/TestSamzaContainer.scala

@@ -24,26 +24,26 @@ import org.junit.Assert._
 import org.junit.Test
 import org.apache.samza.Partition
 import org.apache.samza.config.MapConfig
+import org.apache.samza.metrics.JmxServer
+import org.apache.samza.system.IncomingMessageEnvelope
 import org.apache.samza.system.SystemConsumers
-import org.apache.samza.system.chooser.RoundRobinChooser
 import org.apache.samza.system.SystemConsumer
 import org.apache.samza.system.SystemProducers
 import org.apache.samza.system.SystemProducer
+import org.apache.samza.system.SystemStreamPartition
+import org.apache.samza.system.SystemStream
+import org.apache.samza.system.StreamMetadataCache
+import org.apache.samza.system.chooser.RoundRobinChooser
 import org.apache.samza.serializers.SerdeManager
 import org.apache.samza.task.StreamTask
 import org.apache.samza.task.MessageCollector
-import org.apache.samza.system.IncomingMessageEnvelope
 import org.apache.samza.task.TaskCoordinator
 import org.apache.samza.task.InitableTask
 import org.apache.samza.task.TaskContext
 import org.apache.samza.task.ClosableTask
-import org.apache.samza.system.SystemStreamPartition
-import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin
-import org.apache.samza.system.SystemStream
-import org.apache.samza.system.StreamMetadataCache
 import org.apache.samza.task.TaskInstanceCollector
+import org.apache.samza.util.SinglePartitionWithoutOffsetsSystemAdmin
 import org.scalatest.junit.AssertionsForJUnit
-import org.apache.samza.metrics.JmxServer
 
 class TestSamzaContainer extends AssertionsForJUnit {
   @Test