SAMZA-1990: Samza framework should let using the same system stream as both input and output.

**Symptom:** An `IllegalArgumentException` is thrown when the same `streamId` is referred from multiple input/output stream descriptors.
**Cause:** The `ApplicationDescriptorImpl` caches the serde instances for streams by a `streamId` and there's a check to ensure the expected stream serde matches when using the same stream from multiple input/output descriptors. However the check is incorrect because it compares serde instances and not serde types. This check always fails in this scenario.
**Fix:** Compare the stream serdes for a particular `streamId` by type.

Please take a look prateekm nickpan47
CC: atoomula

Author: Daniel Nishimura <dnishimura@linkedin.com>

Reviewers: prateekm

Closes apache#928 from dnishimura/samza-1990-same-stream-different-inputoutputdescriptors

Author: Daniel Nishimura

samza-core/src/main/java/org/apache/samza/application/descriptors/ApplicationDescriptorImpl.java

@@ -311,9 +311,12 @@ KV<Serde, Serde> getOrCreateStreamSerdes(String streamId, Serde serde) {
             ". Values will not be (de)serialized");
       }
       streamSerdes.put(streamId, KV.of(keySerde, valueSerde));
-    } else if (!currentSerdePair.getKey().equals(keySerde) || !currentSerdePair.getValue().equals(valueSerde)) {
+    } else if (!currentSerdePair.getKey().getClass().equals(keySerde.getClass())
+        || !currentSerdePair.getValue().getClass().equals(valueSerde.getClass())) {
       throw new IllegalArgumentException(String.format("Serde for streamId: %s is already defined. Cannot change it to "
           + "different serdes.", streamId));
+    } else {
+      LOGGER.warn("Using previously defined serde for streamId: " + streamId + ".");
     }
     return streamSerdes.get(streamId);
   }

samza-test/src/test/java/org/apache/samza/test/samzasql/TestSamzaSqlEndToEnd.java

@@ -174,10 +174,7 @@ public void testEndToEndMultiSqlStmts() {
     Assert.assertEquals(numMessages, outMessagesSet.size());
     Assert.assertTrue(IntStream.range(0, numMessages).boxed().collect(Collectors.toList()).equals(new ArrayList<>(outMessagesSet)));
   }
-
-  // The below test won't work until SAMZA-1990 is fixed. Currently, Samza framework does not allow same system stream
-  // to be used as both input and output stream.
-  @Ignore
+  
   @Test
   public void testEndToEndMultiSqlStmtsWithSameSystemStreamAsInputAndOutput() {
     int numMessages = 20;