Skip to content
This repository was archived by the owner on Nov 12, 2024. It is now read-only.

use appropriate topology names #80

Merged
laxmanchekka merged 2 commits into from
Sep 29, 2023
Merged

use appropriate topology names #80

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
74a4d46
use appropriate topology names
laxmanchekka Sep 29, 2023
c87b613
use appropriate topology names
laxmanchekka Sep 29, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion view-generator-framework/build.gradle.kts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ dependencies {
testImplementation("org.junit-pioneer:junit-pioneer:2.0.0")
testImplementation("org.mockito:mockito-core:5.2.0")
testImplementation("org.apache.kafka:kafka-streams-test-utils:7.2.1-ccs")
testRuntimeOnly("org.apache.logging.log4j:log4j-slf4j-impl:2.20.0")
testRuntimeOnly("org.apache.logging.log4j:log4j-slf4j2-impl:2.20.0")
Copy link
Copy Markdown
Contributor

@aaron-steinfeld aaron-steinfeld Sep 29, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the api on line 22 is slf4j1, why are we changing to bind the slf4j2 impl?

Copy link
Copy Markdown
Contributor Author

@laxmanchekka laxmanchekka Sep 29, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Getting this error with slf4j and moving to slf4j2 resolved this.

SLF4J: Defaulting to no-operation (NOP) logger implementation
SLF4J: See https://www.slf4j.org/codes.html#noProviders for further details.
SLF4J: Class path contains SLF4J bindings targeting slf4j-api versions 1.7.x or earlier.
SLF4J: Ignoring binding found at [jar:file:/Users/laxman/.gradle/caches/modules-2/files-2.1/org.apache.logging.log4j/log4j-slf4j-impl/2.20.0/7ab4f082fd162f60afcaf2b8744a3d959feab3e8/log4j-slf4j-impl-2.20.0.jar!/org/slf4j/impl/StaticLoggerBinder.class]
SLF4J: See https://www.slf4j.org/codes.html#ignoredBindings for an explanation.

Copy link
Copy Markdown
Contributor

@aaron-steinfeld aaron-steinfeld Sep 29, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah you're right, my mistake - I'm not sure what I was looking at. Line 22 api declaration is slf4j2 not 1.

}

// Disabling compatibility check for the test avro definitions.
Expand Down
11 changes: 8 additions & 3 deletions ...mework/src/main/java/org/hypertrace/core/viewgenerator/service/ViewGeneratorLauncher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,7 @@ public StreamsBuilder buildTopology(
String outputTopic = jobConfig.getString(OUTPUT_TOPIC_CONFIG_KEY);

KStream<String, Object> mergedStream = null;

int mergedStreamId = 0;
for (String topic : inputTopics) {
KStream<String, Object> inputStream = (KStream<String, Object>) inputStreams.get(topic);

Expand All @@ -64,7 +64,10 @@ public StreamsBuilder buildTopology(
if (mergedStream == null) {
mergedStream = inputStream;
} else {
mergedStream = mergedStream.merge(inputStream, Named.as("merged-stream"));
mergedStream =
mergedStream.merge(
inputStream, Named.as("merged-stream-" + getViewGenName() + "-" + mergedStreamId));
mergedStreamId++;
}
}

Expand All @@ -83,7 +86,9 @@ public StreamsBuilder buildTopology(
}

mergedStream
.process(() -> new ViewGenerationProcessor(getJobConfigKey()))
.process(
() -> new ViewGenerationProcessor(getJobConfigKey()),
Named.as("processor-" + getViewGenName()))
.to(
outputTopic,
Produced.with(
Expand Down
10 changes: 9 additions & 1 deletion ...ework/src/test/java/org/hypertrace/core/viewgenerator/MultiViewGeneratorLauncherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.TestInputTopic;
import org.apache.kafka.streams.TestOutputTopic;
import org.apache.kafka.streams.Topology;
import org.apache.kafka.streams.TopologyTestDriver;
import org.hypertrace.core.kafkastreams.framework.serdes.AvroSerde;
import org.hypertrace.core.serviceframework.config.ConfigClientFactory;
Expand All @@ -29,8 +30,12 @@
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junitpioneer.jupiter.SetEnvironmentVariable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MultiViewGeneratorLauncherTest {
private static final Logger logger =
LoggerFactory.getLogger(MultiViewGeneratorLauncherTest.class);
private MultiViewGeneratorLauncher underTest;
private List<TestInputTopic<String, SpanTypeOne>> inputTopics = new ArrayList<>();
private TestOutputTopic<String, SpanTypeTwo> spanTypeTwoOutputTopic;
Expand All @@ -54,10 +59,11 @@ public void setUp() {
StreamsBuilder streamsBuilder =
underTest.buildTopology(mergedProps, new StreamsBuilder(), new HashMap<>());

Topology testTopology = streamsBuilder.build();
Properties props = new Properties();
mergedProps.forEach(props::put);

TopologyTestDriver td = new TopologyTestDriver(streamsBuilder.build(), props);
TopologyTestDriver td = new TopologyTestDriver(testTopology, props);

Serde<SpanTypeOne> spanTypeOneSerde = new AvroSerde<>();
spanTypeOneSerde.configure(Map.of(), false);
Expand Down Expand Up @@ -87,6 +93,8 @@ public void setUp() {
"test-raw-service-type-output-topic",
new StringSerde().deserializer(),
rawServiceTypeSerde.deserializer());

logger.info("test topology: {}", testTopology.describe());
}

@Test
Expand Down
8 changes: 8 additions & 0 deletions view-generator-framework/src/test/resources/log4j2.properties
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
status=error
name=PropertiesConfig
appender.console.type=Console
appender.console.name=STDOUT
appender.console.layout.type=PatternLayout
appender.console.layout.pattern=%d{yyyy-MM-dd HH:mm:ss.SSS} [%thread] %-5level %c{1.} - %msg%n
rootLogger.level=INFO
rootLogger.appenderRef.stdout.ref=STDOUT