### Question:
SinkNode extends ProcessorNode<K, V> { @Override public void process(final K key, final V value) { final RecordCollector collector = ((RecordCollector.Supplier) context).recordCollector(); final long timestamp = context.timestamp(); if (timestamp < 0) { throw new StreamsException("Invalid (negative) timestamp of " + timestamp + " for output record <" + key + ":" + value + ">."); } try { collector.send(topic, key, value, timestamp, keySerializer, valSerializer, partitioner); } catch (final ClassCastException e) { final String keyClass = key == null ? "unknown because key is null" : key.getClass().getName(); final String valueClass = value == null ? "unknown because value is null" : value.getClass().getName(); throw new StreamsException( String.format("A serializer (key: %s / value: %s) is not compatible to the actual key or value type " + "(key type: %s / value type: %s). Change the default Serdes in StreamConfig or " + "provide correct Serdes via method parameters.", keySerializer.getClass().getName(), valSerializer.getClass().getName(), keyClass, valueClass), e); } } SinkNode(final String name,
                    final String topic,
                    final Serializer<K> keySerializer,
                    final Serializer<V> valSerializer,
                    final StreamPartitioner<? super K, ? super V> partitioner); @Override void addChild(final ProcessorNode<?, ?> child); @SuppressWarnings("unchecked") @Override void init(final ProcessorContext context); @Override void process(final K key, final V value); @Override String toString(); @Override String toString(final String indent);  }### Answer:
@Test @SuppressWarnings("unchecked") public void shouldThrowStreamsExceptionOnInputRecordWithInvalidTimestamp() { final Bytes anyKey = new Bytes("any key".getBytes()); final Bytes anyValue = new Bytes("any value".getBytes()); context.setTime(-1); try { sink.process(anyKey, anyValue); fail("Should have thrown StreamsException"); } catch (final StreamsException ignored) { } }
@Test @SuppressWarnings("unchecked") public void shouldThrowStreamsExceptionOnKeyValueTypeSerializerMismatch() { final String keyOfDifferentTypeThanSerializer = "key with different type"; final String valueOfDifferentTypeThanSerializer = "value with different type"; context.setTime(0); try { sink.process(keyOfDifferentTypeThanSerializer, valueOfDifferentTypeThanSerializer); fail("Should have thrown StreamsException"); } catch (final StreamsException e) { assertThat(e.getCause(), instanceOf(ClassCastException.class)); } }
@Test @SuppressWarnings("unchecked") public void shouldHandleNullKeysWhenThrowingStreamsExceptionOnKeyValueTypeSerializerMismatch() { final String invalidValueToTriggerSerializerMismatch = ""; context.setTime(1); try { sink.process(null, invalidValueToTriggerSerializerMismatch); fail("Should have thrown StreamsException"); } catch (final StreamsException e) { assertThat(e.getCause(), instanceOf(ClassCastException.class)); assertThat(e.getMessage(), containsString("unknown because key is null")); } }
@Test @SuppressWarnings("unchecked") public void shouldHandleNullValuesWhenThrowingStreamsExceptionOnKeyValueTypeSerializerMismatch() { final String invalidKeyToTriggerSerializerMismatch = ""; context.setTime(1); try { sink.process(invalidKeyToTriggerSerializerMismatch, null); fail("Should have thrown StreamsException"); } catch (final StreamsException e) { assertThat(e.getCause(), instanceOf(ClassCastException.class)); assertThat(e.getMessage(), containsString("unknown because value is null")); } }