@Test public void shouldReturnTopicFromRecordContext() throws Exception { assertThat(context.topic(), equalTo(recordContext.topic())); }
@Override public String topic() { if (recordContext == null) { throw new IllegalStateException("This should not happen as topic() should only be called while a record is processed"); } final String topic = recordContext.topic(); if (topic.equals(NONEXIST_TOPIC)) { return null; } return topic; }
AbstractProcessorContext implements InternalProcessorContext { @Override public String topic() { if (recordContext == null) { throw new IllegalStateException("This should not happen as topic() should only be called while a record is processed"); } final String topic = recordContext.topic(); if (topic.equals(NONEXIST_TOPIC)) { return null; } return topic; } }
AbstractProcessorContext implements InternalProcessorContext { @Override public String topic() { if (recordContext == null) { throw new IllegalStateException("This should not happen as topic() should only be called while a record is processed"); } final String topic = recordContext.topic(); if (topic.equals(NONEXIST_TOPIC)) { return null; } return topic; } AbstractProcessorContext(final TaskId taskId,
                             final String applicationId,
                             final StreamsConfig config,
                             final StreamsMetrics metrics,
                             final StateManager stateManager,
                             final ThreadCache cache); }
AbstractProcessorContext implements InternalProcessorContext { @Override public String topic() { if (recordContext == null) { throw new IllegalStateException("This should not happen as topic() should only be called while a record is processed"); } final String topic = recordContext.topic(); if (topic.equals(NONEXIST_TOPIC)) { return null; } return topic; } AbstractProcessorContext(final TaskId taskId,
                             final String applicationId,
                             final StreamsConfig config,
                             final StreamsMetrics metrics,
                             final StateManager stateManager,
                             final ThreadCache cache); @Override String applicationId(); @Override TaskId taskId(); @Override Serde<?> keySerde(); @Override Serde<?> valueSerde(); @Override File stateDir(); @Override StreamsMetrics metrics(); @Override void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback); @Override String topic(); @Override int partition(); @Override long offset(); @Override long timestamp(); @Override Map<String, Object> appConfigs(); @Override Map<String, Object> appConfigsWithPrefix(String prefix); @Override void setRecordContext(final RecordContext recordContext); @Override RecordContext recordContext(); @Override void setCurrentNode(final ProcessorNode currentNode); @Override ProcessorNode currentNode(); @Override ThreadCache getCache(); @Override void initialized(); }
AbstractProcessorContext implements InternalProcessorContext { @Override public String topic() { if (recordContext == null) { throw new IllegalStateException("This should not happen as topic() should only be called while a record is processed"); } final String topic = recordContext.topic(); if (topic.equals(NONEXIST_TOPIC)) { return null; } return topic; } AbstractProcessorContext(final TaskId taskId,
                             final String applicationId,
                             final StreamsConfig config,
                             final StreamsMetrics metrics,
                             final StateManager stateManager,
                             final ThreadCache cache); @Override String applicationId(); @Override TaskId taskId(); @Override Serde<?> keySerde(); @Override Serde<?> valueSerde(); @Override File stateDir(); @Override StreamsMetrics metrics(); @Override void register(final StateStore store, final boolean loggingEnabled, final StateRestoreCallback stateRestoreCallback); @Override String topic(); @Override int partition(); @Override long offset(); @Override long timestamp(); @Override Map<String, Object> appConfigs(); @Override Map<String, Object> appConfigsWithPrefix(String prefix); @Override void setRecordContext(final RecordContext recordContext); @Override RecordContext recordContext(); @Override void setCurrentNode(final ProcessorNode currentNode); @Override ProcessorNode currentNode(); @Override ThreadCache getCache(); @Override void initialized();  }